chore: Adding MCP decorators

azure/functions/decorators/constants.py

@@ -45,3 +45,4 @@
 SEMANTIC_SEARCH = "semanticSearch"
 MYSQL = "mysql"
 MYSQL_TRIGGER = "mysqlTrigger"
+MCP_TOOL_TRIGGER = "mcpToolTrigger"

azure/functions/decorators/function_app.py

@@ -42,6 +42,7 @@
     AssistantQueryInput, AssistantPostInput, InputType, EmbeddingsInput, \
     semantic_search_system_prompt, \
     SemanticSearchInput, EmbeddingsStoreOutput
+from .mcp import MCPToolTrigger
 from .retry_policy import RetryPolicy
 from .function_name import FunctionName
 from .warmup import WarmUpTrigger
@@ -1511,6 +1512,57 @@ def decorator():
 
         return wrap
 
+    def mcp_tool_trigger(self,
+                         arg_name: str,
+                         tool_name: str,
+                         description: Optional[str] = None,
+                         tool_properties: Optional[str] = None,
+                         data_type: Optional[Union[DataType, str]] = None,
+                         **kwargs) -> Callable[..., Any]:
+        """
+        The `mcp_tool_trigger` decorator adds :class:`MCPToolTrigger` to the
+        :class:`FunctionBuilder` object for building a :class:`Function` object
+        used in the worker function indexing model.
+
+        This is equivalent to defining `MCPToolTrigger` in the `function.json`,
+        which enables the function to be triggered when MCP tool requests are
+        received by the host.
+
+        All optional fields will be given default values by the function host when
+        they are parsed.
+
+        Ref: https://aka.ms/remote-mcp-functions-python
+
+        :param arg_name: The name of the trigger parameter in the function code.
+        :param tool_name: The logical tool name exposed to the host.
+        :param description: Optional human-readable description of the tool.
+        :param tool_properties: JSON-serialized tool properties/parameters list.
+        :param data_type: Defines how the Functions runtime should treat the
+            parameter value.
+        :param kwargs: Keyword arguments for specifying additional binding
+            fields to include in the binding JSON.
+
+        :return: Decorator function.
+        """
+
+        @self._configure_function_builder
+        def wrap(fb):
+            def decorator():
+                fb.add_trigger(
+                    trigger=MCPToolTrigger(
+                        name=arg_name,
+                        tool_name=tool_name,
+                        description=description,
+                        tool_properties=tool_properties,
+                        data_type=parse_singular_param_to_enum(data_type,
+                                                               DataType),
+                        **kwargs))
+                return fb
+
+            return decorator()
+
+        return wrap
+
     def dapr_service_invocation_trigger(self,
                                         arg_name: str,
                                         method_name: str,

azure/functions/decorators/mcp.py

@@ -0,0 +1,25 @@
+from typing import Optional
+
+from azure.functions.decorators.constants import (
+    MCP_TOOL_TRIGGER
+)
+from azure.functions.decorators.core import Trigger, DataType
+
+
+class MCPToolTrigger(Trigger):
+
+    @staticmethod
+    def get_binding_name() -> str:
+        return MCP_TOOL_TRIGGER
+
+    def __init__(self,
+                 name: str,
+                 tool_name: str,
+                 description: Optional[str] = None,
+                 tool_properties: Optional[str] = None,
+                 data_type: Optional[DataType] = None,
+                 **kwargs):
+        self.tool_name = tool_name
+        self.description = description
+        self.tool_properties = tool_properties
+        super().__init__(name=name, data_type=data_type)

azure/functions/mcp.py

@@ -0,0 +1,50 @@
+import typing
+
+from . import meta
+
+
+class MCPToolTriggerConverter(meta.InConverter, binding='mcpToolTrigger',
+                              trigger=True):
+
+    @classmethod
+    def check_input_type_annotation(cls, pytype: type) -> bool:
+        return issubclass(pytype, (str, dict, bytes))
+
+    @classmethod
+    def has_implicit_output(cls) -> bool:
+        return True
+
+    @classmethod
+    def decode(cls, data: meta.Datum, *, trigger_metadata):
+        """
+        Decode incoming MCP tool request data.
+        Returns the raw data in its native format (string, dict, bytes).
+        """
+        # Handle different data types appropriately
+        if data.type == 'json':
+            # If it's already parsed JSON, use the value directly
+            return data.value
+        elif data.type == 'string':
+            # If it's a string, use it as-is
+            return data.value
+        elif data.type == 'bytes':
+            return data.value
+        else:
+            # Fallback to python_value for other types
+            return data.python_value if hasattr(data, 'python_value') else data.value
+
+    @classmethod
+    def encode(cls, obj: typing.Any, *, expected_type: typing.Optional[type] = None):
+        """
+        Encode the return value from MCP tool functions.
+        MCP tools typically return string responses.
+        """
+        if obj is None:
+            return meta.Datum(type='string', value='')
+        elif isinstance(obj, str):
+            return meta.Datum(type='string', value=obj)
+        elif isinstance(obj, (bytes, bytearray)):
+            return meta.Datum(type='bytes', value=bytes(obj))
+        else:
+            # Convert other types to string
+            return meta.Datum(type='string', value=str(obj))

tests/decorators/test_mcp.py

@@ -0,0 +1,46 @@
+import unittest
+
+from azure.functions import DataType
+from azure.functions.decorators.core import BindingDirection
+from azure.functions.decorators.mcp import MCPToolTrigger
+from azure.functions.mcp import MCPToolTriggerConverter
+from azure.functions.meta import Datum
+
+
+class TestMCP(unittest.TestCase):
+    def test_mcp_tool_trigger_valid_creation(self):
+        trigger = MCPToolTrigger(
+            name="context",
+            tool_name="hello",
+            description="Hello world.",
+            tool_properties="[]",
+            data_type=DataType.UNDEFINED,
+            dummy_field="dummy",
+        )
+        self.assertEqual(trigger.get_binding_name(), "mcpToolTrigger")
+        self.assertEqual(
+            trigger.get_dict_repr(),
+            {
+                "name": "context",
+                "toolName": "hello",
+                "description": "Hello world.",
+                "toolProperties": "[]",
+                "type": "mcpToolTrigger",
+                "dataType": DataType.UNDEFINED,
+                "dummyField": "dummy",
+                "direction": BindingDirection.IN,
+            },
+        )
+
+    def test_trigger_converter(self):
+        # Test with string data
+        datum = Datum(value='{"arguments":{}}', type='string')
+        result = MCPToolTriggerConverter.decode(datum, trigger_metadata={})
+        self.assertEqual(result, '{"arguments":{}}')
+        self.assertIsInstance(result, str)
+
+        # Test with json data
+        datum_json = Datum(value={"arguments": {}}, type='json')
+        result_json = MCPToolTriggerConverter.decode(datum_json, trigger_metadata={})
+        self.assertEqual(result_json, {"arguments": {}})
+        self.assertIsInstance(result_json, dict)