feat: add register_converter API

azure/functions/__init__.py

@@ -23,7 +23,7 @@
 from ._http_asgi import AsgiMiddleware
 from .kafka import KafkaEvent, KafkaConverter, KafkaTriggerConverter
 from .mcp import MCPToolContext, PromptInvocationContext
-from .meta import get_binding_registry
+from .meta import get_binding_registry, register_converter
 from ._queue import QueueMessage
 from ._servicebus import ServiceBusMessage
 from ._sql import SqlRow, SqlRowList
@@ -51,6 +51,7 @@
 __all__ = (
     # Functions
     'get_binding_registry',
+    'register_converter',
 
     # Generics.
     'Context',

azure/functions/meta.py

@@ -15,6 +15,16 @@
 )
 
 
+# Binding names whose converters are permitted to be overridden via
+# register_converter(). Only durable-related bindings are overridable.
+_OVERRIDABLE_BINDINGS: frozenset = frozenset({
+    'orchestrationTrigger',
+    'entityTrigger',
+    'activityTrigger',
+    'durableClient',
+})
+
+
 def is_iterable_type_annotation(annotation: object, pytype: object) -> bool:
     is_iterable_anno = (
         typing_inspect.is_generic_type(annotation)
@@ -407,3 +417,47 @@ def encode(cls, obj: Any, *,
 
 def get_binding_registry():
     return _ConverterMeta
+
+
+def register_converter(
+        binding_name: str,
+        converter_cls: type,
+        *,
+        overwrite: bool = False) -> None:
+    """Register or replace a converter for a binding name.
+
+    By default raises RuntimeError if the binding is already registered,
+    requiring callers to explicitly pass overwrite=True to replace an
+    existing entry. This API is intended for integration packages (e.g.
+    azure-functions-durable) that need to override built-in durable
+    converters without accessing private internals.
+
+    Parameters
+    ----------
+    binding_name:
+        The binding type string as it appears in function.json, e.g.
+        ``'orchestrationTrigger'``.
+    converter_cls:
+        A class that is a subclass of InConverter and/or OutConverter.
+    overwrite:
+        If True, silently replace any existing registration for
+        ``binding_name``. If False (default), raise RuntimeError when the
+        binding is already registered.
+    """
+    if not isinstance(converter_cls, type):
+        raise TypeError('converter_cls must be a class')
+    if not issubclass(converter_cls, (InConverter, OutConverter)):
+        raise TypeError(
+            'converter_cls must be a subclass of InConverter and/or '
+            'OutConverter')
+    if binding_name not in _OVERRIDABLE_BINDINGS:
+        raise ValueError(
+            f'cannot register converter for {binding_name!r}: '
+            f'register_converter() only supports durable-related bindings. '
+            f'Overridable bindings: {sorted(_OVERRIDABLE_BINDINGS)}')
+    if not overwrite and binding_name in _ConverterMeta._bindings:
+        raise RuntimeError(
+            f'cannot register a converter for {binding_name!r} binding: '
+            f'another converter for this binding has already been '
+            f'registered. Pass overwrite=True to replace it.')
+    _ConverterMeta._bindings[binding_name] = converter_cls

tests/test_meta.py

@@ -254,3 +254,100 @@ def _parse_datetime(self, datetime_str):
 
     def _parse_timedelta(self, timedelta_str):
         return meta._BaseConverter._parse_timedelta(timedelta_str)
+
+
+class TestRegisterConverter(unittest.TestCase):
+
+    def setUp(self):
+        # Snapshot the live bindings registry so each test is isolated.
+        self._saved_bindings = dict(meta._ConverterMeta._bindings)
+
+    def tearDown(self):
+        # Restore the registry exactly as it was before the test ran.
+        meta._ConverterMeta._bindings.clear()
+        meta._ConverterMeta._bindings.update(self._saved_bindings)
+
+    @staticmethod
+    def _make_dummy_converter():
+        """Return a fresh converter class usable as a placeholder."""
+        class _Dummy(meta.InConverter, binding=None):
+            @classmethod
+            def check_input_type_annotation(cls, pytype: type) -> bool:
+                return True
+
+            @classmethod
+            def decode(cls, data: meta.Datum, *, trigger_metadata):
+                return None
+
+            @classmethod
+            def has_implicit_output(cls) -> bool:
+                return False
+
+        return _Dummy
+
+    # ------------------------------------------------------------------ #
+    # Allow-list enforcement                                               #
+    # ------------------------------------------------------------------ #
+
+    def test_non_durable_binding_raises_value_error(self):
+        with self.assertRaises(ValueError) as ctx:
+            meta.register_converter('httpTrigger',
+                                    self._make_dummy_converter())
+        self.assertIn('httpTrigger', str(ctx.exception))
+
+    def test_non_durable_binding_with_overwrite_still_raises_value_error(self):
+        # overwrite=True must not bypass the allow-list.
+        with self.assertRaises(ValueError):
+            meta.register_converter('queueTrigger',
+                                    self._make_dummy_converter(),
+                                    overwrite=True)
+
+    # ------------------------------------------------------------------ #
+    # Successful registration                                              #
+    # ------------------------------------------------------------------ #
+
+    def test_all_durable_bindings_are_overridable(self):
+        for name in ('orchestrationTrigger', 'entityTrigger',
+                     'activityTrigger', 'durableClient'):
+            dummy = self._make_dummy_converter()
+            meta.register_converter(name, dummy, overwrite=True)
+            self.assertIs(
+                meta._ConverterMeta._bindings[name], dummy,
+                msg=f'{name!r} was not registered correctly')
+
+    def test_register_new_durable_binding_not_previously_in_registry(self):
+        meta._ConverterMeta._bindings.pop('orchestrationTrigger', None)
+        dummy = self._make_dummy_converter()
+        meta.register_converter('orchestrationTrigger', dummy)
+        self.assertIs(meta._ConverterMeta._bindings['orchestrationTrigger'],
+                      dummy)
+
+    # ------------------------------------------------------------------ #
+    # overwrite=False guard                                                #
+    # ------------------------------------------------------------------ #
+
+    def test_already_registered_without_overwrite_raises_runtime_error(self):
+        dummy = self._make_dummy_converter()
+        meta._ConverterMeta._bindings['orchestrationTrigger'] = dummy
+        with self.assertRaises(RuntimeError) as ctx:
+            meta.register_converter('orchestrationTrigger',
+                                    self._make_dummy_converter())
+        self.assertIn('orchestrationTrigger', str(ctx.exception))
+
+    def test_already_registered_with_overwrite_true_replaces_entry(self):
+        original = self._make_dummy_converter()
+        replacement = self._make_dummy_converter()
+        meta._ConverterMeta._bindings['orchestrationTrigger'] = original
+        meta.register_converter('orchestrationTrigger', replacement,
+                                overwrite=True)
+        self.assertIs(meta._ConverterMeta._bindings['orchestrationTrigger'],
+                      replacement)
+
+    # ------------------------------------------------------------------ #
+    # Public API surface                                                   #
+    # ------------------------------------------------------------------ #
+
+    def test_register_converter_is_exported_from_azure_functions(self):
+        import azure.functions as af
+        self.assertTrue(hasattr(af, 'register_converter'))
+        self.assertIn('register_converter', af.__all__)