From cc8caee4a0904790c10282dc278d0f5f9c6b2756 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Tue, 21 Apr 2026 11:25:56 -0500 Subject: [PATCH 1/3] add support for using official mcp sdk types --- azure/functions/__init__.py | 10 +- azure/functions/decorators/function_app.py | 236 +++++--- azure/functions/mcp.py | 215 ++++---- pyproject.toml | 3 +- tests/decorators/test_mcp.py | 601 ++++++--------------- 5 files changed, 452 insertions(+), 613 deletions(-) diff --git a/azure/functions/__init__.py b/azure/functions/__init__.py index 6eaae640..9eff7eb4 100644 --- a/azure/functions/__init__.py +++ b/azure/functions/__init__.py @@ -21,8 +21,7 @@ from ._http_wsgi import WsgiMiddleware from ._http_asgi import AsgiMiddleware from .kafka import KafkaEvent, KafkaConverter, KafkaTriggerConverter -from .mcp import (MCPToolContext, PromptInvocationContext, ContentBlock, TextContentBlock, - ImageContentBlock, ResourceLinkBlock, CallToolResult) +from .mcp import MCPToolContext, PromptInvocationContext from .meta import get_binding_registry from ._queue import QueueMessage from ._servicebus import ServiceBusMessage @@ -111,13 +110,6 @@ 'PromptArgument', 'McpPropertyType', 'mcp_content', - - # MCP ContentBlock types - 'ContentBlock', - 'TextContentBlock', - 'ImageContentBlock', - 'ResourceLinkBlock', - 'CallToolResult' ) __version__ = '2.2.0b1' diff --git a/azure/functions/decorators/function_app.py b/azure/functions/decorators/function_app.py index 189eb7ee..2f48dbf7 100644 --- a/azure/functions/decorators/function_app.py +++ b/azure/functions/decorators/function_app.py @@ -51,16 +51,18 @@ _SemanticSearchInput, _EmbeddingsStoreOutput from .mcp import _MCPToolTrigger, MCPResourceTrigger, MCPPromptTrigger, \ PromptArgument, build_property_metadata, \ - has_mcp_content_marker, should_create_structured_content + has_mcp_content_marker from .retry_policy import RetryPolicy from .function_name import FunctionName from .warmup import WarmUpTrigger -from ..mcp import MCPToolContext, ContentBlock, CallToolResult +from ..mcp import MCPToolContext, _is_mcp_call_tool_result, _is_mcp_sdk_type from .._http_asgi import AsgiMiddleware from .._http_wsgi import WsgiMiddleware, Context from azure.functions.decorators.mysql import MySqlInput, MySqlOutput, \ MySqlTrigger +_logger = logging.getLogger('azure.functions.AsgiMiddleware') + class Function(object): """ @@ -1633,36 +1635,46 @@ def decorator(fb: FunctionBuilder) -> FunctionBuilder: return_annotation = sig.return_annotation if return_annotation != inspect.Signature.empty and not auto_use_result_schema: - # Check if return type is a ContentBlock subclass - is_content_block = False - is_call_tool_result = False + # Check if return type is an MCP SDK type or @mcp_content decorated is_mcp_content = False + is_mcp_sdk_type = False + + # Try to detect official MCP SDK types by checking module + try: + if isinstance(return_annotation, type): + # Check if the type is from the mcp.types module + if hasattr(return_annotation, '__module__'): + module = return_annotation.__module__ + if module and (module.startswith('mcp.types') or module == 'mcp.types'): + is_mcp_sdk_type = True + except (ImportError, TypeError, AttributeError): + pass + # Check for @mcp_content decorated classes try: - # Handle direct ContentBlock or CallToolResult if isinstance(return_annotation, type): - if issubclass(return_annotation, ContentBlock): - is_content_block = True - elif issubclass(return_annotation, CallToolResult): - is_call_tool_result = True - elif has_mcp_content_marker(return_annotation): + if has_mcp_content_marker(return_annotation): is_mcp_content = True except TypeError: pass - # Handle List[ContentBlock] and other generic types + # Handle List[MCP types] and Optional[MCP types] if hasattr(return_annotation, '__origin__'): import typing origin = typing.get_origin(return_annotation) args = typing.get_args(return_annotation) - # Check for List[ContentBlock] or list[ContentBlock] - if origin in (list, List) and args: - try: - if issubclass(args[0], ContentBlock): - is_content_block = True - except TypeError: - pass + # Check for official MCP SDK types in lists + try: + if isinstance(args[0], type): + # Check if the type is from the mcp.types module + if hasattr(args[0], '__module__'): + module = args[0].__module__ + if module and (module.startswith('mcp.types') + or module == 'mcp.types'): + is_mcp_sdk_type = True + except (ImportError, TypeError, AttributeError): + pass # Check for Optional[T] where T is an MCP type if origin is Union: @@ -1671,20 +1683,33 @@ def decorator(fb: FunctionBuilder) -> FunctionBuilder: continue try: if isinstance(arg, type): - if issubclass(arg, ContentBlock): - is_content_block = True - break - elif issubclass(arg, CallToolResult): - is_call_tool_result = True - break - elif has_mcp_content_marker(arg): + if has_mcp_content_marker(arg): is_mcp_content = True break except TypeError: pass + # Check for MCP SDK types in Union + try: + if isinstance(arg, type): + # Check module for mcp.types + if hasattr(arg, '__module__'): + module = arg.__module__ + if (module + and (module.startswith('mcp.types') + or module == 'mcp.types')): + is_mcp_sdk_type = True + break + except (ImportError, TypeError, AttributeError): + pass + # Auto-enable use_result_schema for MCP types - if is_content_block or is_call_tool_result or is_mcp_content: + if is_mcp_content or is_mcp_sdk_type: + _logger.info( + f"Auto-detected MCP content return type for " + f"function '{target_func.__name__}'. " + f"Setting use_result_schema=True for proper " + f"structured content handling.") auto_use_result_schema = True # Pull any explicitly declared MCP tool properties @@ -1737,63 +1762,126 @@ async def wrapper(context: str, *args, **kwargs): if result is None: return "" - # Handle CallToolResult - manual construction by user - if isinstance(result, CallToolResult): - result_dict = result.to_dict() - structured = (json.dumps(result.structured_content) - if result.structured_content else None) - return json.dumps({ - "type": "call_tool_result", - "content": json.dumps(result_dict), - "structuredContent": structured - }) - - # Handle List[ContentBlock] - multiple content blocks - if isinstance(result, list) and all( - isinstance(item, ContentBlock) for item in result): - content_blocks = [block.to_dict() for block in result] - return json.dumps({ - "type": "multi_content_result", - "content": json.dumps(content_blocks), - "structuredContent": json.dumps(content_blocks) - }) - - # Handle single ContentBlock - if isinstance(result, ContentBlock): - block_dict = result.to_dict() - return str(json.dumps({ - "type": result.type, - "content": json.dumps(block_dict), - "structuredContent": json.dumps(block_dict) - })) - # Handle structured content generation when # auto_use_result_schema is True if auto_use_result_schema: - # Check if we should create structured content - if should_create_structured_content(result): - # Serialize result as JSON for structured content - # Handle dataclasses properly - if dataclasses.is_dataclass(result): - result_json = json.dumps( - dataclasses.asdict(result)) - elif hasattr(result, '__dict__'): - # For regular classes with __dict__ - result_json = json.dumps(result.__dict__) + _logger.info( + f"Processing result of function " + f"'{target_func.__name__}' for structured content " + f"generation.") + + # Handle official MCP SDK CallToolResult + if _is_mcp_call_tool_result(result): + # CallToolResult already has the correct structure + # Serialize using model_dump() for Pydantic models + if hasattr(result, 'model_dump'): + result_dict = result.model_dump( + mode='json', exclude_none=True) + elif hasattr(result, 'dict'): + result_dict = result.dict(exclude_none=True) + else: + # Fallback: convert to dict manually + result_dict = { + 'content': [ + block.model_dump( + mode='json', exclude_none=True) + if hasattr(block, 'model_dump') + else dict(block) + for block in result.content + ] if hasattr(result, 'content') else [] + } + if (hasattr(result, 'structuredContent') + and result.structuredContent is not None): + result_dict['structuredContent'] = ( + result.structuredContent) + + # Full CallToolResult as JSON string + full_result_json = json.dumps(result_dict) + + # Extract structuredContent value + structured_content_value = result_dict.get( + 'structuredContent') + structured_content_json = json.dumps( + structured_content_value + ) if structured_content_value is not None else None + + # Return in the expected format for CallToolResult + return str(json.dumps({ + "type": "call_tool_result", + "content": full_result_json, + "structuredContent": structured_content_json + })) + + # Handle all other MCP SDK types + # (TextContent, ImageContent, ResourceLink, etc.) + elif _is_mcp_sdk_type(result): + if hasattr(result, 'model_dump'): + result_dict = result.model_dump( + mode='json', exclude_none=True) + result_json = json.dumps(result_dict) + elif hasattr(result, 'dict'): + result_dict = result.dict(exclude_none=True) + result_json = json.dumps(result_dict) else: - # Fallback to str conversion - result_json = json.dumps( - result) if not isinstance( - result, str) else result + result_dict = result.__dict__ + result_json = json.dumps(result_dict) - # Return McpToolResult format with both text and structured content + # Extract type from the object itself + # (works for any MCP type) + # e.g., TextContent has type="text", + result_type = result_dict.get('type', 'text') + + # Return format with type extracted from the object + return str(json.dumps({ + "type": result_type, + "content": result_json, + "structuredContent": result_json + })) + + # Handle dataclasses (e.g., @mcp_content decorated) + elif dataclasses.is_dataclass(result): + result_json = json.dumps( + dataclasses.asdict(result)) + + # Return format with both text and structured content + return str(json.dumps({ + "type": "text", + "content": json.dumps( + {"type": "text", "text": result_json}), + "structuredContent": result_json + })) + + # Handle regular classes with __dict__ + elif hasattr(result, '__dict__'): + result_json = json.dumps(result.__dict__) + + # Return format with both text and structured content return str(json.dumps({ "type": "text", - "content": json.dumps({"type": "text", "text": result_json}), + "content": json.dumps( + {"type": "text", "text": result_json}), "structuredContent": result_json })) + else: + # Fallback to str conversion + result_json = json.dumps( + result) if not isinstance( + result, str) else result - return str(result) + # Return format with both text and structured content + return str(json.dumps({ + "type": "text", + "content": json.dumps( + {"type": "text", "text": result_json}), + "structuredContent": result_json + })) + else: + # Backwards compatibility: when structured content + # is not enabled, return the result as-is + if isinstance(result, str): + return result + else: + return str(result) wrapper.__signature__ = wrapper_sig fb._function._func = wrapper diff --git a/azure/functions/mcp.py b/azure/functions/mcp.py index 4c22400c..cc79bd17 100644 --- a/azure/functions/mcp.py +++ b/azure/functions/mcp.py @@ -1,10 +1,29 @@ # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. import typing -from dataclasses import dataclass, field, asdict -from typing import Optional, List, Any +import json +from dataclasses import asdict, is_dataclass +from typing import Any from . import meta +import logging + +_logger = logging.getLogger('azure.functions.AsgiMiddleware') + + +# Try to import the official MCP SDK types if available +# This allows users to use the canonical types without us taking a hard dependency +_MCP_SDK_AVAILABLE = False +_mcp_types = None + +try: + _logger.info("Attempting to import official MCP SDK types for better compatibility.") + from mcp import types as _mcp_types + _MCP_SDK_AVAILABLE = True +except ImportError: + _logger.warning("Official MCP SDK not found. Using fallback types. " + "For best compatibility with MCP tools, please install the 'mcp' package.") + _mcp_types = None # MCP-specific context object @@ -14,90 +33,65 @@ class MCPToolContext(typing.Dict[str, typing.Any]): pass -# ContentBlock types for MCP responses -@dataclass -class ContentBlock: - """Base class for MCP content blocks.""" - type: str = field(init=False) - - def to_dict(self) -> dict: - """Convert the content block to a dictionary for JSON serialization.""" - return asdict(self) - - -@dataclass -class TextContentBlock(ContentBlock): - """Text content block for MCP responses.""" - text: str - type: str = field(default="text", init=False) - - -@dataclass -class ImageContentBlock(ContentBlock): - """Image content block for MCP responses.""" - data: str # base64-encoded image data - mime_type: str - type: str = field(default="image", init=False) - - def to_dict(self) -> dict: - """Convert to dict with correct JSON property names.""" - return { - "type": self.type, - "data": self.data, - "mimeType": self.mime_type - } - - -@dataclass -class ResourceLinkBlock(ContentBlock): - """Resource link content block for MCP responses.""" - uri: str - name: Optional[str] = None - description: Optional[str] = None - mime_type: Optional[str] = None - type: str = field(default="resource_link", init=False) - - def to_dict(self) -> dict: - """Convert to dict with correct JSON property names.""" - result = { - "type": self.type, - "uri": self.uri - } - if self.name is not None: - result["name"] = self.name - if self.description is not None: - result["description"] = self.description - if self.mime_type is not None: - result["mimeType"] = self.mime_type - return result - - -@dataclass -class CallToolResult: +def _is_mcp_sdk_type(obj: Any) -> bool: + """Check if an object is from the official MCP SDK. + + This uses module checking to detect any MCP SDK type, + avoiding the need to hard-code specific type names. """ - Result type for MCP tool calls that allows manual construction - of content blocks and structured content. + if not _MCP_SDK_AVAILABLE or _mcp_types is None: + return False - Example: - return CallToolResult( - content=[ - TextContentBlock(text="Here's the data"), - ImageContentBlock(data=base64_data, mime_type="image/png") - ], - structured_content={"key": "value"} - ) + # Check if the object's class is from the mcp.types module + obj_type = type(obj) + if hasattr(obj_type, '__module__'): + module = obj_type.__module__ + # Check if it's from mcp.types or any mcp submodule + if module and (module.startswith('mcp.types') or module == 'mcp.types'): + return True + + return False + + +def _is_mcp_call_tool_result(obj: Any) -> bool: + """Check if an object is CallToolResult from the official MCP SDK.""" + if not _MCP_SDK_AVAILABLE or _mcp_types is None: + return False + + # Check for CallToolResult from mcp.types + if hasattr(_mcp_types, 'CallToolResult'): + return isinstance(obj, _mcp_types.CallToolResult) + + return False + + +def _serialize_content_block(block: Any) -> dict: + """Serialize a content block to a dictionary. + + Handles official mcp.types and @mcp_content decorated classes. """ - content: List[ContentBlock] - structured_content: Optional[Any] = None + # If it's from the official MCP SDK + if _is_mcp_sdk_type(block): + # MCP SDK types should be JSON-serializable + if hasattr(block, 'model_dump'): + return block.model_dump(mode='json', exclude_none=True) + elif hasattr(block, 'dict'): + return block.dict(exclude_none=True) + + # If it's a dataclass (e.g., @mcp_content decorated) + if is_dataclass(block) and not isinstance(block, type): + return asdict(block) + + # If it's already a dict + if isinstance(block, dict): + return block - def to_dict(self) -> dict: - """Convert to dictionary for JSON serialization.""" - result = { - "content": [block.to_dict() for block in self.content] - } - if self.structured_content is not None: - result["structuredContent"] = self.structured_content - return result + # Fallback: try to convert to dict + if hasattr(block, '__dict__'): + return block.__dict__ + + msg = f"Unable to serialize content block of type {type(block).__name__}" + raise TypeError(msg) class PromptInvocationContext: @@ -167,8 +161,8 @@ def has_implicit_output(cls) -> bool: @classmethod def decode(cls, data: meta.Datum, *, trigger_metadata): - """ - Decode incoming MCP tool request data. + """Decode incoming MCP tool request data. + Returns the raw data in its native format (string, dict, bytes). """ # Handle different data types appropriately @@ -185,10 +179,15 @@ def decode(cls, data: meta.Datum, *, trigger_metadata): 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. + def encode(cls, obj: typing.Any, *, + expected_type: typing.Optional[type] = None): + """Encode the return value from MCP tool functions. + + Supports multiple return types: + - Strings, bytes + - Official mcp.types content blocks (TextContent, ImageContent, etc.) + - Official mcp.types.CallToolResult + - Classes decorated with @mcp_content """ if obj is None: return meta.Datum(type='string', value='') @@ -196,9 +195,43 @@ def encode(cls, obj: typing.Any, *, expected_type: typing.Optional[type] = None) 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)) + + # Handle official MCP SDK CallToolResult + elif _is_mcp_call_tool_result(obj): + # Serialize the MCP SDK's CallToolResult + if hasattr(obj, 'model_dump'): + result_dict = obj.model_dump(mode='json', exclude_none=True) + elif hasattr(obj, 'dict'): + result_dict = obj.dict(exclude_none=True) + else: + # Fallback: try to access attributes directly + content_blocks = [_serialize_content_block(block) + for block in obj.content] + result_dict = { + 'content': content_blocks if hasattr(obj, 'content') else [], + } + if (hasattr(obj, 'structured_content') + and obj.structured_content is not None): + result_dict['structuredContent'] = obj.structured_content + result_json = json.dumps(result_dict) + return meta.Datum(type='string', value=result_json) + + # Handle official MCP SDK content types + elif _is_mcp_sdk_type(obj): + serialized = _serialize_content_block(obj) + return meta.Datum(type='string', value=json.dumps(serialized)) + + # Handle list of MCP SDK content blocks + elif isinstance(obj, list) and len(obj) > 0: + # Check if it's a list of MCP SDK content blocks + first_item = obj[0] + if _is_mcp_sdk_type(first_item): + blocks_list = [_serialize_content_block(block) + for block in obj] + return meta.Datum(type='string', value=json.dumps(blocks_list)) + + # Fallback: convert other types to string + return meta.Datum(type='string', value=str(obj)) class MCPResourceTriggerConverter(meta.InConverter, binding='mcpResourceTrigger', diff --git a/pyproject.toml b/pyproject.toml index ad3f9ae4..42149240 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,7 +37,8 @@ dev = [ 'azure-functions-durable', 'flake8~=4.0.1; python_version < "3.11"', 'flake8~=7.1.1; python_version >= "3.11"', - 'flake8-docstrings' + 'flake8-docstrings', + 'mcp' ] [tool.setuptools.packages.find] diff --git a/tests/decorators/test_mcp.py b/tests/decorators/test_mcp.py index 65ecb4f1..5533069a 100644 --- a/tests/decorators/test_mcp.py +++ b/tests/decorators/test_mcp.py @@ -3,6 +3,10 @@ import typing import unittest import json +import asyncio +from dataclasses import dataclass +from unittest.mock import patch +from typing import List, Optional import azure.functions as func from azure.functions import (DataType, MCPToolContext, @@ -14,6 +18,12 @@ from azure.functions.mcp import (_MCPToolTriggerConverter, MCPResourceTriggerConverter) from azure.functions.meta import Datum +from mcp.types import ( + ResourceLink, + TextContent, + ImageContent, + CallToolResult +) class TestMCP(unittest.TestCase): @@ -596,357 +606,6 @@ class DataModel: self.assertTrue(hasattr(DataModel, '__mcp_content__')) -class TestContentBlocks(unittest.TestCase): - """Tests for ContentBlock types""" - - def test_text_content_block_creation(self): - """Test creating a TextContentBlock""" - block = func.TextContentBlock(text="Hello, world!") - self.assertEqual(block.type, "text") - self.assertEqual(block.text, "Hello, world!") - self.assertEqual(block.to_dict(), {"type": "text", "text": "Hello, world!"}) - - def test_image_content_block_creation(self): - """Test creating an ImageContentBlock""" - block = func.ImageContentBlock(data="base64data", mime_type="image/png") - self.assertEqual(block.type, "image") - self.assertEqual(block.data, "base64data") - self.assertEqual(block.mime_type, "image/png") - - block_dict = block.to_dict() - self.assertEqual(block_dict["type"], "image") - self.assertEqual(block_dict["data"], "base64data") - self.assertEqual(block_dict["mimeType"], "image/png") - - def test_resource_link_block_creation(self): - """Test creating a ResourceLinkBlock""" - block = func.ResourceLinkBlock( - uri="https://example.com/resource", - name="Example Resource", - description="A test resource", - mime_type="application/json" - ) - self.assertEqual(block.type, "resource_link") - self.assertEqual(block.uri, "https://example.com/resource") - self.assertEqual(block.name, "Example Resource") - - block_dict = block.to_dict() - self.assertEqual(block_dict["type"], "resource_link") - self.assertEqual(block_dict["uri"], "https://example.com/resource") - self.assertEqual(block_dict["mimeType"], "application/json") - - def test_resource_link_block_minimal(self): - """Test ResourceLinkBlock with only required fields""" - block = func.ResourceLinkBlock(uri="file://logo.png") - self.assertEqual(block.type, "resource_link") - self.assertEqual(block.uri, "file://logo.png") - - block_dict = block.to_dict() - self.assertEqual(block_dict["type"], "resource_link") - self.assertEqual(block_dict["uri"], "file://logo.png") - self.assertNotIn("name", block_dict) - self.assertNotIn("description", block_dict) - - def test_call_tool_result_creation(self): - """Test creating a CallToolResult""" - result = func.CallToolResult( - content=[ - func.TextContentBlock(text="Here's the data"), - func.ImageContentBlock(data="imagedata", mime_type="image/jpeg") - ], - structured_content={"key": "value", "count": 42} - ) - - self.assertEqual(len(result.content), 2) - self.assertIsInstance(result.content[0], func.TextContentBlock) - self.assertIsInstance(result.content[1], func.ImageContentBlock) - self.assertEqual(result.structured_content, {"key": "value", "count": 42}) - - result_dict = result.to_dict() - self.assertIn("content", result_dict) - self.assertIn("structuredContent", result_dict) - self.assertEqual(len(result_dict["content"]), 2) - - def test_call_tool_result_without_structured_content(self): - """Test CallToolResult without structured content""" - result = func.CallToolResult( - content=[func.TextContentBlock(text="Simple text")] - ) - - self.assertIsNone(result.structured_content) - result_dict = result.to_dict() - self.assertIn("content", result_dict) - self.assertEqual(result.structured_content, None) - - def test_text_content_block_empty_string(self): - """Test TextContentBlock with empty string""" - block = func.TextContentBlock(text="") - self.assertEqual(block.text, "") - self.assertEqual(block.to_dict(), {"type": "text", "text": ""}) - - def test_text_content_block_multiline(self): - """Test TextContentBlock with multiline text""" - multiline_text = """Line 1 -Line 2 -Line 3""" - block = func.TextContentBlock(text=multiline_text) - self.assertEqual(block.text, multiline_text) - block_dict = block.to_dict() - self.assertIn("Line 1\nLine 2\nLine 3", block_dict["text"]) - - def test_text_content_block_special_characters(self): - """Test TextContentBlock with special characters""" - special_text = 'Text with "quotes" and \'apostrophes\' and ' - block = func.TextContentBlock(text=special_text) - self.assertEqual(block.text, special_text) - block_dict = block.to_dict() - self.assertEqual(block_dict["text"], special_text) - - def test_image_content_block_different_mime_types(self): - """Test ImageContentBlock with various MIME types""" - mime_types = ["image/png", "image/jpeg", "image/gif", "image/svg+xml"] - for mime_type in mime_types: - block = func.ImageContentBlock(data="data123", mime_type=mime_type) - self.assertEqual(block.mime_type, mime_type) - block_dict = block.to_dict() - self.assertEqual(block_dict["mimeType"], mime_type) - - def test_image_content_block_property_naming(self): - """Test that ImageContentBlock uses camelCase in JSON (mimeType not mime_type)""" - block = func.ImageContentBlock(data="base64", mime_type="image/png") - block_dict = block.to_dict() - - # Should use camelCase in JSON - self.assertIn("mimeType", block_dict) - self.assertNotIn("mime_type", block_dict) - self.assertEqual(block_dict["mimeType"], "image/png") - - def test_image_content_block_large_data(self): - """Test ImageContentBlock with large base64 data""" - large_data = "A" * 10000 # Simulate large base64 string - block = func.ImageContentBlock(data=large_data, mime_type="image/png") - self.assertEqual(len(block.data), 10000) - block_dict = block.to_dict() - self.assertEqual(len(block_dict["data"]), 10000) - - def test_resource_link_block_all_fields(self): - """Test ResourceLinkBlock with all fields populated""" - block = func.ResourceLinkBlock( - uri="https://example.com/api/resource", - name="Test Resource", - description="A detailed description", - mime_type="application/json" - ) - block_dict = block.to_dict() - - self.assertEqual(block_dict["type"], "resource_link") - self.assertEqual(block_dict["uri"], "https://example.com/api/resource") - self.assertEqual(block_dict["name"], "Test Resource") - self.assertEqual(block_dict["description"], "A detailed description") - self.assertEqual(block_dict["mimeType"], "application/json") - - def test_resource_link_block_partial_fields(self): - """Test ResourceLinkBlock with some optional fields None""" - block = func.ResourceLinkBlock( - uri="file://path/to/file.txt", - name="MyFile" - ) - block_dict = block.to_dict() - - self.assertEqual(block_dict["uri"], "file://path/to/file.txt") - self.assertEqual(block_dict["name"], "MyFile") - self.assertNotIn("description", block_dict) - self.assertNotIn("mimeType", block_dict) - - def test_resource_link_block_file_uri(self): - """Test ResourceLinkBlock with file:// URI""" - block = func.ResourceLinkBlock(uri="file://logo.png") - self.assertEqual(block.uri, "file://logo.png") - block_dict = block.to_dict() - self.assertEqual(block_dict["uri"], "file://logo.png") - - def test_resource_link_block_http_uri(self): - """Test ResourceLinkBlock with http:// and https:// URIs""" - http_block = func.ResourceLinkBlock(uri="http://example.com") - https_block = func.ResourceLinkBlock(uri="https://example.com") - - self.assertEqual(http_block.uri, "http://example.com") - self.assertEqual(https_block.uri, "https://example.com") - - def test_call_tool_result_multiple_text_blocks(self): - """Test CallToolResult with multiple TextContentBlocks""" - result = func.CallToolResult( - content=[ - func.TextContentBlock(text="First paragraph"), - func.TextContentBlock(text="Second paragraph"), - func.TextContentBlock(text="Third paragraph") - ] - ) - - self.assertEqual(len(result.content), 3) - result_dict = result.to_dict() - self.assertEqual(len(result_dict["content"]), 3) - self.assertEqual(result_dict["content"][0]["text"], "First paragraph") - self.assertEqual(result_dict["content"][2]["text"], "Third paragraph") - - def test_call_tool_result_mixed_content_blocks(self): - """Test CallToolResult with mixed ContentBlock types""" - result = func.CallToolResult( - content=[ - func.TextContentBlock(text="Description"), - func.ResourceLinkBlock(uri="https://link.com", name="Link"), - func.ImageContentBlock(data="img123", mime_type="image/png"), - func.TextContentBlock(text="Footer") - ] - ) - - self.assertEqual(len(result.content), 4) - result_dict = result.to_dict() - - # Verify each block is correctly serialized - self.assertEqual(result_dict["content"][0]["type"], "text") - self.assertEqual(result_dict["content"][1]["type"], "resource_link") - self.assertEqual(result_dict["content"][2]["type"], "image") - self.assertEqual(result_dict["content"][3]["type"], "text") - - def test_call_tool_result_structured_content_dict(self): - """Test CallToolResult with dict structured_content""" - metadata = { - "id": "123", - "name": "Test", - "tags": ["tag1", "tag2"], - "count": 42 - } - - result = func.CallToolResult( - content=[func.TextContentBlock(text="Data")], - structured_content=metadata - ) - - result_dict = result.to_dict() - self.assertEqual(result_dict["structuredContent"], metadata) - self.assertEqual(result_dict["structuredContent"]["id"], "123") - self.assertEqual(result_dict["structuredContent"]["count"], 42) - - def test_call_tool_result_structured_content_nested(self): - """Test CallToolResult with nested structured_content""" - nested_data = { - "user": { - "id": 1, - "name": "John", - "profile": { - "age": 30, - "location": "NYC" - } - }, - "metadata": { - "timestamp": "2026-03-18T00:00:00Z" - } - } - - result = func.CallToolResult( - content=[func.TextContentBlock(text="User data")], - structured_content=nested_data - ) - - result_dict = result.to_dict() - self.assertEqual(result_dict["structuredContent"]["user"]["name"], "John") - self.assertEqual(result_dict["structuredContent"]["user"]["profile"]["age"], 30) - - def test_call_tool_result_structured_content_list(self): - """Test CallToolResult with list as structured_content""" - list_data = [ - {"id": 1, "name": "Item 1"}, - {"id": 2, "name": "Item 2"}, - {"id": 3, "name": "Item 3"} - ] - - result = func.CallToolResult( - content=[func.TextContentBlock(text="Items")], - structured_content=list_data - ) - - result_dict = result.to_dict() - self.assertIsInstance(result_dict["structuredContent"], list) - self.assertEqual(len(result_dict["structuredContent"]), 3) - self.assertEqual(result_dict["structuredContent"][1]["name"], "Item 2") - - def test_call_tool_result_empty_content_list(self): - """Test CallToolResult with empty content list""" - result = func.CallToolResult(content=[]) - - self.assertEqual(len(result.content), 0) - result_dict = result.to_dict() - self.assertEqual(result_dict["content"], []) - - def test_content_blocks_json_serialization(self): - """Test that ContentBlocks can be JSON serialized""" - import json - - blocks = [ - func.TextContentBlock(text="Hello"), - func.ImageContentBlock(data="base64", mime_type="image/png"), - func.ResourceLinkBlock(uri="https://example.com") - ] - - # Convert to dicts and serialize - blocks_dict = [block.to_dict() for block in blocks] - json_str = json.dumps(blocks_dict) - - # Verify it's valid JSON - parsed = json.loads(json_str) - self.assertEqual(len(parsed), 3) - self.assertEqual(parsed[0]["type"], "text") - self.assertEqual(parsed[1]["mimeType"], "image/png") - - def test_call_tool_result_json_serialization(self): - """Test that CallToolResult can be JSON serialized""" - import json - - result = func.CallToolResult( - content=[ - func.TextContentBlock(text="Test"), - func.ImageContentBlock(data="abc123", mime_type="image/jpeg") - ], - structured_content={"key": "value", "number": 123} - ) - - result_dict = result.to_dict() - json_str = json.dumps(result_dict) - - # Verify it's valid JSON - parsed = json.loads(json_str) - self.assertIn("content", parsed) - self.assertIn("structuredContent", parsed) - self.assertEqual(parsed["structuredContent"]["key"], "value") - - def test_content_block_inheritance(self): - """Test that all ContentBlock types inherit from ContentBlock""" - text_block = func.TextContentBlock(text="test") - image_block = func.ImageContentBlock(data="data", mime_type="image/png") - resource_block = func.ResourceLinkBlock(uri="uri") - - self.assertIsInstance(text_block, func.ContentBlock) - self.assertIsInstance(image_block, func.ContentBlock) - self.assertIsInstance(resource_block, func.ContentBlock) - - def test_content_block_type_immutable(self): - """Test that type field is set correctly and consistently""" - text_block = func.TextContentBlock(text="test") - image_block = func.ImageContentBlock(data="data", mime_type="image/png") - resource_block = func.ResourceLinkBlock(uri="uri") - - # Type should be set via field(init=False) - self.assertEqual(text_block.type, "text") - self.assertEqual(image_block.type, "image") - self.assertEqual(resource_block.type, "resource_link") - - # Verify in dict output - self.assertEqual(text_block.to_dict()["type"], "text") - self.assertEqual(image_block.to_dict()["type"], "image") - self.assertEqual(resource_block.to_dict()["type"], "resource_link") - - class TestAutoUseResultSchema(unittest.TestCase): """Tests for automatic use_result_schema detection""" @@ -956,76 +615,70 @@ def setUp(self): def tearDown(self): self.app = None - def test_auto_detect_resource_link_block(self): - """Test auto-detection of ResourceLinkBlock return type""" + def test_auto_detect_mcp_resource_link(self): + """Test auto-detection of MCP SDK ResourceLink return type""" @self.app.mcp_tool() - def get_logo() -> func.ResourceLinkBlock: + def get_logo() -> ResourceLink: """Returns a logo""" - return func.ResourceLinkBlock(uri="file://logo.png") + return ResourceLink( + type="resource_link", + uri="file://logo.png", + name="Logo" + ) trigger = get_logo._function._bindings[0] self.assertTrue(trigger.use_result_schema) - def test_auto_detect_text_content_block(self): - """Test auto-detection of TextContentBlock return type""" + def test_auto_detect_mcp_text_content(self): + """Test auto-detection of MCP SDK TextContent return type""" @self.app.mcp_tool() - def get_text() -> func.TextContentBlock: + def get_text() -> TextContent: """Returns text""" - return func.TextContentBlock(text="Hello") + return TextContent(type="text", text="Hello") trigger = get_text._function._bindings[0] self.assertTrue(trigger.use_result_schema) - def test_auto_detect_image_content_block(self): - """Test auto-detection of ImageContentBlock return type""" + def test_auto_detect_mcp_image_content(self): + """Test auto-detection of MCP SDK ImageContent return type""" @self.app.mcp_tool() - def get_image() -> func.ImageContentBlock: + def get_image() -> ImageContent: """Returns image""" - return func.ImageContentBlock(data="base64", mime_type="image/png") + return ImageContent( + type="image", + data="base64data", + mimeType="image/png" + ) trigger = get_image._function._bindings[0] self.assertTrue(trigger.use_result_schema) - def test_auto_detect_call_tool_result(self): - """Test auto-detection of CallToolResult return type""" + def test_auto_detect_mcp_call_tool_result(self): + """Test auto-detection of MCP SDK CallToolResult return type""" @self.app.mcp_tool() - def get_result() -> func.CallToolResult: + def get_result() -> CallToolResult: """Returns CallToolResult""" - return func.CallToolResult(content=[]) + return CallToolResult( + content=[TextContent(type="text", text="result")] + ) trigger = get_result._function._bindings[0] self.assertTrue(trigger.use_result_schema) - def test_auto_detect_list_content_block(self): - """Test auto-detection of List[ContentBlock] return type""" - from typing import List - + def test_auto_detect_list_mcp_text_content(self): + """Test auto-detection of List[TextContent] return type""" @self.app.mcp_tool() - def get_multiple() -> List[func.ContentBlock]: - """Returns multiple blocks""" - return [func.TextContentBlock(text="test")] - - trigger = get_multiple._function._bindings[0] - self.assertTrue(trigger.use_result_schema) - - def test_auto_detect_list_text_content_block(self): - """Test auto-detection of List[TextContentBlock] return type""" - from typing import List - - @self.app.mcp_tool() - def get_texts() -> List[func.TextContentBlock]: + def get_texts() -> List[TextContent]: """Returns text blocks""" - return [func.TextContentBlock(text="test")] + return [TextContent(type="text", text="test")] trigger = get_texts._function._bindings[0] self.assertTrue(trigger.use_result_schema) - def test_auto_detect_optional_content_block(self): - """Test auto-detection of Optional[ContentBlock] return type""" - from typing import Optional - + def test_auto_detect_optional_mcp_image_content(self): + """Test auto-detection of Optional[ImageContent] return type""" @self.app.mcp_tool() - def maybe_image() -> Optional[func.ImageContentBlock]: + def maybe_image() -> Optional[ImageContent]: """Maybe returns image""" return None @@ -1107,19 +760,9 @@ def explicit_false() -> str: trigger = explicit_false._function._bindings[0] self.assertFalse(trigger.use_result_schema) - def test_explicit_overrides_auto_detection(self): - """Test that explicit value is not overridden by auto-detection""" - @self.app.mcp_tool(use_result_schema=True) - def override_test() -> func.ResourceLinkBlock: - """Override test""" - return func.ResourceLinkBlock(uri="test") - - trigger = override_test._function._bindings[0] - self.assertTrue(trigger.use_result_schema) - class TestStructuredContentInResponses(unittest.TestCase): - """Tests for structuredContent field in MCP responses""" + """Tests for structuredContent field in MCP responses with official MCP SDK types""" def setUp(self): self.app = func.FunctionApp() @@ -1128,16 +771,13 @@ def tearDown(self): self.app = None def test_structured_content_in_call_tool_result(self): - """Test that CallToolResult includes structuredContent""" - import json - import asyncio - + """Test that MCP SDK CallToolResult includes structuredContent""" @self.app.mcp_tool() - def test_func() -> func.CallToolResult: + def test_func() -> CallToolResult: """Test function""" - return func.CallToolResult( - content=[func.TextContentBlock(text="test")], - structured_content={"key": "value"} + return CallToolResult( + content=[TextContent(type="text", text="test")], + structuredContent={"key": "value"} ) # Get the wrapper function @@ -1157,22 +797,31 @@ def test_func() -> func.CallToolResult: self.assertEqual(result_obj["type"], "call_tool_result") self.assertIsNotNone(result_obj["structuredContent"]) - def test_structured_content_in_single_content_block(self): - """Test that single ContentBlock includes structuredContent""" - import json - import asyncio + # Verify structuredContent value + structured_obj = json.loads(result_obj["structuredContent"]) + self.assertEqual(structured_obj, {"key": "value"}) + def test_structured_content_in_resource_link(self): + """Test that MCP SDK ResourceLink includes structuredContent""" @self.app.mcp_tool() - def test_func() -> func.ResourceLinkBlock: + def test_func() -> ResourceLink: """Test function""" - return func.ResourceLinkBlock(uri="file://test.png", name="Test") + return ResourceLink( + type="resource_link", + uri="file://test.png", + name="Test", + mimeType="image/png" + ) wrapper = test_func._function._func context = json.dumps({"arguments": {}}) result = asyncio.run(wrapper(context)) result_obj = json.loads(result) + self.assertIn("type", result_obj) + self.assertIn("content", result_obj) self.assertIn("structuredContent", result_obj) + self.assertEqual(result_obj["type"], "resource_link") self.assertIsNotNone(result_obj["structuredContent"]) # Verify structuredContent matches content @@ -1180,54 +829,44 @@ def test_func() -> func.ResourceLinkBlock: structured_obj = json.loads(result_obj["structuredContent"]) self.assertEqual(content_obj, structured_obj) - def test_structured_content_in_list_content_blocks(self): - """Test that List[ContentBlock] includes structuredContent""" - import json - import asyncio - from typing import List - + def test_structured_content_in_text_content(self): + """Test that MCP SDK TextContent includes structuredContent""" @self.app.mcp_tool() - def test_func() -> List[func.ContentBlock]: + def test_func() -> TextContent: """Test function""" - return [ - func.TextContentBlock(text="First"), - func.TextContentBlock(text="Second") - ] + return TextContent(type="text", text="Hello World") wrapper = test_func._function._func context = json.dumps({"arguments": {}}) result = asyncio.run(wrapper(context)) result_obj = json.loads(result) + self.assertIn("type", result_obj) + self.assertEqual(result_obj["type"], "text") + self.assertIn("content", result_obj) self.assertIn("structuredContent", result_obj) self.assertIsNotNone(result_obj["structuredContent"]) - # Verify structuredContent matches content - content_obj = json.loads(result_obj["content"]) - structured_obj = json.loads(result_obj["structuredContent"]) - self.assertEqual(content_obj, structured_obj) - def test_structured_content_with_mcp_content_decorator(self): """Test that @mcp_content decorated class includes structuredContent""" - import json - import asyncio - @func.mcp_content + @dataclass class MyData: - def __init__(self, name: str, value: int): - self.name = name - self.value = value + name: str + value: int @self.app.mcp_tool() def test_func() -> MyData: """Test function""" - return MyData("test", 42) + return MyData(name="test", value=42) wrapper = test_func._function._func context = json.dumps({"arguments": {}}) result = asyncio.run(wrapper(context)) result_obj = json.loads(result) + self.assertIn("type", result_obj) + self.assertIn("content", result_obj) self.assertIn("structuredContent", result_obj) self.assertIsNotNone(result_obj["structuredContent"]) @@ -1236,6 +875,92 @@ def test_func() -> MyData: self.assertEqual(structured_obj["name"], "test") self.assertEqual(structured_obj["value"], 42) + def test_backwards_compatibility_string_without_use_result_schema(self): + """Test that plain string returns work without use_result_schema""" + @self.app.mcp_tool() + def test_func() -> str: + """Test function""" + return "Hello!" + + wrapper = test_func._function._func + context = json.dumps({"arguments": {}}) + result = asyncio.run(wrapper(context)) + + # Should return the raw string, not a JSON structure + self.assertEqual(result, "Hello!") + self.assertIsInstance(result, str) + + def test_explicit_use_result_schema_with_string(self): + """Test that explicit use_result_schema=True structures string response""" + @self.app.mcp_tool(use_result_schema=True) + def test_func() -> str: + """Test function""" + return "Hello!" + + wrapper = test_func._function._func + context = json.dumps({"arguments": {}}) + result = asyncio.run(wrapper(context)) + + # Should return structured JSON + result_obj = json.loads(result) + self.assertIn("type", result_obj) + self.assertIn("content", result_obj) + self.assertIn("structuredContent", result_obj) + + +class TestMCPPackageNotInstalled(unittest.TestCase): + """Tests for graceful degradation when mcp package is not installed""" + + def setUp(self): + self.app = func.FunctionApp() + + def tearDown(self): + self.app = None + + def test_no_auto_detect_when_mcp_not_installed(self): + """Test that auto-detection doesn't happen when mcp package is not available""" + # Mock sys.modules to simulate mcp not being installed + import sys + with patch.dict(sys.modules, {'mcp': None, 'mcp.types': None}): + # Clear any cached imports + import importlib + if 'azure.functions.decorators.function_app' in sys.modules: + importlib.reload(sys.modules['azure.functions.decorators.function_app']) + + # Create a new app after mocking + test_app = func.FunctionApp() + + @test_app.mcp_tool() + def get_data() -> str: + """Returns data""" + return "test" + + trigger = get_data._function._bindings[0] + # Should not auto-detect when mcp is not available + self.assertFalse(trigger.use_result_schema) + + def test_mcp_content_decorator_still_works_without_mcp(self): + """Test that @mcp_content decorator works even when mcp package is not installed""" + @func.mcp_content + class MyData: + def __init__(self, value: str): + self.value = value + + # Decorator should still mark the class + self.assertTrue(hasattr(MyData, '__mcp_content__')) + self.assertEqual(MyData.__mcp_content__, True) + + def test_explicit_use_result_schema_works_without_mcp(self): + """Test that explicit use_result_schema=True works without mcp package""" + @self.app.mcp_tool(use_result_schema=True) + def test_func() -> str: + """Test function""" + return "Hello!" + + trigger = test_func._function._bindings[0] + # Explicit parameter should always work + self.assertTrue(trigger.use_result_schema) + class TestPromptArgument(unittest.TestCase): """Unit tests for PromptArgument dataclass""" From 436d4751e7299004b6e4df3d4dbb866d0978024c Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Tue, 21 Apr 2026 11:45:13 -0500 Subject: [PATCH 2/3] remove logger --- azure/functions/decorators/function_app.py | 11 ----------- azure/functions/mcp.py | 6 ------ 2 files changed, 17 deletions(-) diff --git a/azure/functions/decorators/function_app.py b/azure/functions/decorators/function_app.py index 2f48dbf7..a1a91e24 100644 --- a/azure/functions/decorators/function_app.py +++ b/azure/functions/decorators/function_app.py @@ -61,8 +61,6 @@ from azure.functions.decorators.mysql import MySqlInput, MySqlOutput, \ MySqlTrigger -_logger = logging.getLogger('azure.functions.AsgiMiddleware') - class Function(object): """ @@ -1705,11 +1703,6 @@ def decorator(fb: FunctionBuilder) -> FunctionBuilder: # Auto-enable use_result_schema for MCP types if is_mcp_content or is_mcp_sdk_type: - _logger.info( - f"Auto-detected MCP content return type for " - f"function '{target_func.__name__}'. " - f"Setting use_result_schema=True for proper " - f"structured content handling.") auto_use_result_schema = True # Pull any explicitly declared MCP tool properties @@ -1765,10 +1758,6 @@ async def wrapper(context: str, *args, **kwargs): # Handle structured content generation when # auto_use_result_schema is True if auto_use_result_schema: - _logger.info( - f"Processing result of function " - f"'{target_func.__name__}' for structured content " - f"generation.") # Handle official MCP SDK CallToolResult if _is_mcp_call_tool_result(result): diff --git a/azure/functions/mcp.py b/azure/functions/mcp.py index cc79bd17..cf51cb83 100644 --- a/azure/functions/mcp.py +++ b/azure/functions/mcp.py @@ -6,9 +6,6 @@ from typing import Any from . import meta -import logging - -_logger = logging.getLogger('azure.functions.AsgiMiddleware') # Try to import the official MCP SDK types if available @@ -17,12 +14,9 @@ _mcp_types = None try: - _logger.info("Attempting to import official MCP SDK types for better compatibility.") from mcp import types as _mcp_types _MCP_SDK_AVAILABLE = True except ImportError: - _logger.warning("Official MCP SDK not found. Using fallback types. " - "For best compatibility with MCP tools, please install the 'mcp' package.") _mcp_types = None From 22a9d6bf7793fec68a15dd6efadcae17c7ed37a5 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Tue, 21 Apr 2026 12:02:31 -0500 Subject: [PATCH 3/3] update import statement --- azure/functions/mcp.py | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/azure/functions/mcp.py b/azure/functions/mcp.py index cf51cb83..10d0761b 100644 --- a/azure/functions/mcp.py +++ b/azure/functions/mcp.py @@ -13,12 +13,6 @@ _MCP_SDK_AVAILABLE = False _mcp_types = None -try: - from mcp import types as _mcp_types - _MCP_SDK_AVAILABLE = True -except ImportError: - _mcp_types = None - # MCP-specific context object class MCPToolContext(typing.Dict[str, typing.Any]): @@ -33,8 +27,14 @@ def _is_mcp_sdk_type(obj: Any) -> bool: This uses module checking to detect any MCP SDK type, avoiding the need to hard-code specific type names. """ + global _MCP_SDK_AVAILABLE, _mcp_types if not _MCP_SDK_AVAILABLE or _mcp_types is None: - return False + try: + from mcp import types as _mcp_types + _MCP_SDK_AVAILABLE = True + except ImportError: + _mcp_types = None + return False # Check if the object's class is from the mcp.types module obj_type = type(obj) @@ -49,8 +49,14 @@ def _is_mcp_sdk_type(obj: Any) -> bool: def _is_mcp_call_tool_result(obj: Any) -> bool: """Check if an object is CallToolResult from the official MCP SDK.""" + global _MCP_SDK_AVAILABLE, _mcp_types if not _MCP_SDK_AVAILABLE or _mcp_types is None: - return False + try: + from mcp import types as _mcp_types + _MCP_SDK_AVAILABLE = True + except ImportError: + _mcp_types = None + return False # Check for CallToolResult from mcp.types if hasattr(_mcp_types, 'CallToolResult'):