diff --git a/azure/functions/__init__.py b/azure/functions/__init__.py index e3ffe6cf..1e12dff4 100644 --- a/azure/functions/__init__.py +++ b/azure/functions/__init__.py @@ -11,6 +11,7 @@ Cardinality, AccessRights, HttpMethod, AsgiFunctionApp, WsgiFunctionApp, ExternalHttpFunctionApp, BlobSource, McpPropertyType) +from .decorators.mcp import mcp_content from ._durable_functions import OrchestrationContext, EntityContext from .decorators.function_app import (FunctionRegister, TriggerApi, BindingApi, SettingsApi) @@ -19,7 +20,8 @@ from ._http_wsgi import WsgiMiddleware from ._http_asgi import AsgiMiddleware from .kafka import KafkaEvent, KafkaConverter, KafkaTriggerConverter -from .mcp import MCPToolContext +from .mcp import (MCPToolContext, ContentBlock, TextContentBlock, + ImageContentBlock, ResourceLinkBlock, CallToolResult) from .meta import get_binding_registry from ._queue import QueueMessage from ._servicebus import ServiceBusMessage @@ -104,7 +106,15 @@ 'HttpMethod', 'BlobSource', 'MCPToolContext', - 'McpPropertyType' + 'McpPropertyType', + 'mcp_content', + + # MCP ContentBlock types + 'ContentBlock', + 'TextContentBlock', + 'ImageContentBlock', + 'ResourceLinkBlock', + 'CallToolResult' ) __version__ = '1.25.0b4' diff --git a/azure/functions/decorators/function_app.py b/azure/functions/decorators/function_app.py index a603c0de..f3196a95 100644 --- a/azure/functions/decorators/function_app.py +++ b/azure/functions/decorators/function_app.py @@ -2,6 +2,7 @@ # Licensed under the MIT License. import abc import asyncio +import dataclasses import functools import inspect import json @@ -48,11 +49,12 @@ _AssistantQueryInput, _AssistantPostInput, InputType, _EmbeddingsInput, \ semantic_search_system_prompt, \ _SemanticSearchInput, _EmbeddingsStoreOutput -from .mcp import _MCPToolTrigger, MCPResourceTrigger, build_property_metadata +from .mcp import _MCPToolTrigger, MCPResourceTrigger, build_property_metadata, \ + has_mcp_content_marker, should_create_structured_content from .retry_policy import RetryPolicy from .function_name import FunctionName from .warmup import WarmUpTrigger -from ..mcp import MCPToolContext +from ..mcp import MCPToolContext, ContentBlock, CallToolResult from .._http_asgi import AsgiMiddleware from .._http_wsgi import WsgiMiddleware, Context from azure.functions.decorators.mysql import MySqlInput, MySqlOutput, \ @@ -1604,7 +1606,7 @@ def decorator(): return wrap - def mcp_tool(self, metadata: Optional[str] = None): + def mcp_tool(self, metadata: Optional[str] = None, use_result_schema: Optional[bool] = False): """Decorator to register an MCP tool function. Ref: https://aka.ms/remote-mcp-functions-python @@ -1615,12 +1617,75 @@ def mcp_tool(self, metadata: Optional[str] = None): - Handles MCPToolContext injection :param metadata: JSON-serialized metadata object for the tool. + :param use_result_schema: Whether the result schema should be + provided by the worker instead of being generated by the host + extension. """ @self._configure_function_builder def decorator(fb: FunctionBuilder) -> FunctionBuilder: target_func = fb._function.get_user_function() sig = inspect.signature(target_func) + # Auto-detect MCP return types and set use_result_schema=True + # Use a separate variable to avoid UnboundLocalError + auto_use_result_schema = use_result_schema + 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 + is_mcp_content = False + + 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): + is_mcp_content = True + except TypeError: + pass + + # Handle List[ContentBlock] and other generic 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 Optional[T] where T is an MCP type + if origin is Union: + for arg in args: + if isinstance(arg, type(None)): + 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): + is_mcp_content = True + break + except TypeError: + pass + + # Auto-enable use_result_schema for MCP types + if is_content_block or is_call_tool_result or is_mcp_content: + auto_use_result_schema = True + # Pull any explicitly declared MCP tool properties explicit_properties = getattr(target_func, "__mcp_tool_properties__", {}) @@ -1667,6 +1732,66 @@ async def wrapper(context: str, *args, **kwargs): result = target_func(**call_kwargs) if asyncio.iscoroutine(result): result = await result + + 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__) + else: + # Fallback to str conversion + result_json = json.dumps( + result) if not isinstance( + result, str) else result + + # Return McpToolResult format with both text and structured content + return str(json.dumps({ + "type": "text", + "content": json.dumps({"type": "text", "text": result_json}), + "structuredContent": result_json + })) + return str(result) wrapper.__signature__ = wrapper_sig @@ -1679,7 +1804,8 @@ async def wrapper(context: str, *args, **kwargs): tool_name=tool_name, description=description, tool_properties=tool_properties_json, - metadata=metadata + metadata=metadata, + use_result_schema=auto_use_result_schema ) ) return fb diff --git a/azure/functions/decorators/mcp.py b/azure/functions/decorators/mcp.py index 435b8864..0c1208f3 100644 --- a/azure/functions/decorators/mcp.py +++ b/azure/functions/decorators/mcp.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. import inspect +import typing from typing import List, Optional, Union, get_origin, get_args from datetime import datetime @@ -37,6 +38,7 @@ def __init__(self, mime_type: Optional[str] = None, size: Optional[int] = None, metadata: Optional[str] = None, + use_result_schema: Optional[bool] = False, data_type: Optional[DataType] = None, **kwargs): self.uri = uri @@ -46,6 +48,7 @@ def __init__(self, self.mimeType = mime_type self.size = size self.metadata = metadata + self.useResultSchema = use_result_schema super().__init__(name=name, data_type=data_type) @@ -61,12 +64,14 @@ def __init__(self, description: Optional[str] = None, tool_properties: Optional[str] = None, metadata: Optional[str] = None, + use_result_schema: Optional[bool] = False, data_type: Optional[DataType] = None, **kwargs): self.tool_name = tool_name self.description = description self.tool_properties = tool_properties self.metadata = metadata + self.use_result_schema = use_result_schema super().__init__(name=name, data_type=data_type) @@ -125,6 +130,7 @@ def check_is_required(param: type, param_type_hint: type) -> bool: def build_property_metadata(sig, skip_param_names: List[str], explicit_properties: dict) -> List[dict]: + """Build the tool_properties list for MCPToolTrigger based on function signature.""" tool_properties = [] for param_name, param in sig.parameters.items(): if param_name in skip_param_names: @@ -156,3 +162,70 @@ def build_property_metadata(sig, tool_properties.append(property_data) return tool_properties + + +def has_mcp_content_marker(obj: typing.Any) -> bool: + """ + Check if an object or its type is marked for structured content generation. + Returns True if the object's class has '__mcp_content__' attribute set to True. + Handles both class types and instances. + """ + if obj is None: + return False + + # If obj is already a class type, check it directly + if isinstance(obj, type): + return getattr(obj, '__mcp_content__', False) is True + + # Otherwise, get the type and check + obj_type = type(obj) + return getattr(obj_type, '__mcp_content__', False) is True + + +def should_create_structured_content(obj: typing.Any) -> bool: + """ + Determines whether structured content should be created for the given object. + + Returns True if: + - The object's class is decorated with a marker that sets __mcp_content__ = True + - The object is not a primitive type (str, int, float, bool, None) + - The object is not a dict or list + + This mimics the .NET implementation's McpContentAttribute checking. + """ + if obj is None: + return False + + # Primitive types don't generate structured content unless explicitly marked + if isinstance(obj, (str, int, float, bool, dict, list)): + return False + + # Check for the marker attribute + return has_mcp_content_marker(obj) + + +def mcp_content(cls): + """ + Decorator to mark a class as an MCP result type that should be serialized + as structured content. + + When a function returns an object of a type decorated with this decorator, + the result will be serialized as both text content (for backwards compatibility) + and structured content (for clients that support it). + + This is the Python equivalent of C#'s [McpContent] attribute. + + Example: + @mcp_content + class ImageMetadata: + def __init__(self, image_id: str, format: str, tags: list): + self.image_id = image_id + self.format = format + self.tags = tags + + @app.mcp_tool(use_result_schema=True) + def get_image_info(): + return ImageMetadata("logo", "png", ["functions"]) + """ + cls.__mcp_content__ = True + return cls diff --git a/azure/functions/mcp.py b/azure/functions/mcp.py index e6e49162..f204ff5c 100644 --- a/azure/functions/mcp.py +++ b/azure/functions/mcp.py @@ -1,6 +1,8 @@ # 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 from . import meta @@ -12,6 +14,92 @@ 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: + """ + Result type for MCP tool calls that allows manual construction + of content blocks and structured content. + + Example: + return CallToolResult( + content=[ + TextContentBlock(text="Here's the data"), + ImageContentBlock(data=base64_data, mime_type="image/png") + ], + structured_content={"key": "value"} + ) + """ + content: List[ContentBlock] + structured_content: Optional[Any] = None + + 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 + + class _MCPToolTriggerConverter(meta.InConverter, binding='mcpToolTrigger', trigger=True): diff --git a/tests/decorators/test_mcp.py b/tests/decorators/test_mcp.py index 2cc315f3..3237e5fe 100644 --- a/tests/decorators/test_mcp.py +++ b/tests/decorators/test_mcp.py @@ -19,6 +19,7 @@ def test_mcp_tool_trigger_valid_creation(self): description="Hello world.", tool_properties="[]", metadata='{"key": "value"}', + use_result_schema=True, data_type=DataType.UNDEFINED, dummy_field="dummy", ) @@ -34,6 +35,7 @@ def test_mcp_tool_trigger_valid_creation(self): "dataType": DataType.UNDEFINED, "dummyField": "dummy", "metadata": '{"key": "value"}', + 'useResultSchema': True, "direction": BindingDirection.IN, }, ) @@ -504,3 +506,727 @@ def test_trigger_converter(self): result_json = MCPResourceTriggerConverter.decode(datum_json, trigger_metadata={}) self.assertEqual(result_json, {"arguments": {}}) self.assertIsInstance(result_json, dict) + + +class TestStructuredContent(unittest.TestCase): + """Tests for structured content functionality""" + + def setUp(self): + self.app = func.FunctionApp() + + def tearDown(self): + self.app = None + + def test_mcp_content_decorator(self): + """Test that @mcp_content decorator marks a class properly""" + from azure.functions.decorators.mcp import has_mcp_content_marker + + @func.mcp_content + class TestData: + def __init__(self, value: str): + self.value = value + + instance = TestData("test") + self.assertTrue(has_mcp_content_marker(instance)) + self.assertTrue(hasattr(TestData, '__mcp_content__')) + self.assertEqual(TestData.__mcp_content__, True) + + def test_should_create_structured_content_for_marked_class(self): + """Test that marked classes generate structured content""" + from azure.functions.decorators.mcp import should_create_structured_content + + @func.mcp_content + class MarkedData: + def __init__(self, name: str): + self.name = name + + instance = MarkedData("test") + self.assertTrue(should_create_structured_content(instance)) + + def test_should_not_create_structured_content_for_primitives(self): + """Test that primitive types don't generate structured content""" + from azure.functions.decorators.mcp import should_create_structured_content + + self.assertFalse(should_create_structured_content("string")) + self.assertFalse(should_create_structured_content(42)) + self.assertFalse(should_create_structured_content(3.14)) + self.assertFalse(should_create_structured_content(True)) + self.assertFalse(should_create_structured_content(None)) + + def test_should_not_create_structured_content_for_unmarked_class(self): + """Test that unmarked classes don't generate structured content""" + from azure.functions.decorators.mcp import should_create_structured_content + + class UnmarkedData: + def __init__(self, value: str): + self.value = value + + instance = UnmarkedData("test") + self.assertFalse(should_create_structured_content(instance)) + + def test_mcp_tool_with_use_result_schema_parameter(self): + """Test that use_result_schema parameter is passed to trigger""" + @self.app.mcp_tool(use_result_schema=True) + def test_tool(value: str): + """Test tool with result schema""" + return value + + trigger = test_tool._function._bindings[0] + self.assertEqual(trigger.use_result_schema, True) + self.assertEqual(trigger.tool_name, "test_tool") + + def test_mcp_content_with_dataclass(self): + """Test mcp_content decorator works with dataclasses""" + from dataclasses import dataclass + from azure.functions.decorators.mcp import should_create_structured_content + + @func.mcp_content + @dataclass + class DataModel: + name: str + count: int + + instance = DataModel(name="test", count=5) + self.assertTrue(should_create_structured_content(instance)) + 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""" + + def setUp(self): + self.app = func.FunctionApp() + + def tearDown(self): + self.app = None + + def test_auto_detect_resource_link_block(self): + """Test auto-detection of ResourceLinkBlock return type""" + @self.app.mcp_tool() + def get_logo() -> func.ResourceLinkBlock: + """Returns a logo""" + return func.ResourceLinkBlock(uri="file://logo.png") + + 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""" + @self.app.mcp_tool() + def get_text() -> func.TextContentBlock: + """Returns text""" + return func.TextContentBlock(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""" + @self.app.mcp_tool() + def get_image() -> func.ImageContentBlock: + """Returns image""" + return func.ImageContentBlock(data="base64", mime_type="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""" + @self.app.mcp_tool() + def get_result() -> func.CallToolResult: + """Returns CallToolResult""" + return func.CallToolResult(content=[]) + + 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 + + @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]: + """Returns text blocks""" + return [func.TextContentBlock(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 + + @self.app.mcp_tool() + def maybe_image() -> Optional[func.ImageContentBlock]: + """Maybe returns image""" + return None + + trigger = maybe_image._function._bindings[0] + self.assertTrue(trigger.use_result_schema) + + def test_auto_detect_mcp_content_class(self): + """Test auto-detection of @mcp_content decorated class""" + @func.mcp_content + class MyData: + def __init__(self, value: str): + self.value = value + + @self.app.mcp_tool() + def get_data() -> MyData: + """Returns custom data""" + return MyData("test") + + trigger = get_data._function._bindings[0] + self.assertTrue(trigger.use_result_schema) + + def test_no_auto_detect_string(self): + """Test that plain string return type doesn't trigger auto-detection""" + @self.app.mcp_tool() + def get_string() -> str: + """Returns string""" + return "Hello" + + trigger = get_string._function._bindings[0] + self.assertFalse(trigger.use_result_schema) + + def test_no_auto_detect_int(self): + """Test that int return type doesn't trigger auto-detection""" + @self.app.mcp_tool() + def get_number() -> int: + """Returns number""" + return 42 + + trigger = get_number._function._bindings[0] + self.assertFalse(trigger.use_result_schema) + + def test_no_auto_detect_dict(self): + """Test that dict return type doesn't trigger auto-detection""" + @self.app.mcp_tool() + def get_dict() -> dict: + """Returns dict""" + return {"key": "value"} + + trigger = get_dict._function._bindings[0] + self.assertFalse(trigger.use_result_schema) + + def test_no_auto_detect_no_annotation(self): + """Test that no return annotation doesn't trigger auto-detection""" + @self.app.mcp_tool() + def no_annotation(): + """No annotation""" + return "test" + + trigger = no_annotation._function._bindings[0] + self.assertFalse(trigger.use_result_schema) + + def test_explicit_use_result_schema_true(self): + """Test that explicit use_result_schema=True is preserved""" + @self.app.mcp_tool(use_result_schema=True) + def explicit_true() -> str: + """Explicit True""" + return "test" + + trigger = explicit_true._function._bindings[0] + self.assertTrue(trigger.use_result_schema) + + def test_explicit_use_result_schema_false(self): + """Test that explicit use_result_schema=False works""" + @self.app.mcp_tool(use_result_schema=False) + def explicit_false() -> str: + """Explicit False""" + return "test" + + 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""" + + def setUp(self): + self.app = func.FunctionApp() + + def tearDown(self): + self.app = None + + def test_structured_content_in_call_tool_result(self): + """Test that CallToolResult includes structuredContent""" + import json + import asyncio + + @self.app.mcp_tool() + def test_func() -> func.CallToolResult: + """Test function""" + return func.CallToolResult( + content=[func.TextContentBlock(text="test")], + structured_content={"key": "value"} + ) + + # Get the wrapper function + wrapper = test_func._function._func + + # Call the wrapper + context = json.dumps({"arguments": {}}) + result = asyncio.run(wrapper(context)) + + # Parse the result + result_obj = json.loads(result) + + # Verify structure + self.assertIn("type", result_obj) + self.assertIn("content", result_obj) + self.assertIn("structuredContent", result_obj) + 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 + + @self.app.mcp_tool() + def test_func() -> func.ResourceLinkBlock: + """Test function""" + return func.ResourceLinkBlock(uri="file://test.png", name="Test") + + wrapper = test_func._function._func + context = json.dumps({"arguments": {}}) + result = asyncio.run(wrapper(context)) + result_obj = json.loads(result) + + 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_in_list_content_blocks(self): + """Test that List[ContentBlock] includes structuredContent""" + import json + import asyncio + from typing import List + + @self.app.mcp_tool() + def test_func() -> List[func.ContentBlock]: + """Test function""" + return [ + func.TextContentBlock(text="First"), + func.TextContentBlock(text="Second") + ] + + wrapper = test_func._function._func + context = json.dumps({"arguments": {}}) + result = asyncio.run(wrapper(context)) + result_obj = json.loads(result) + + 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 + class MyData: + def __init__(self, name: str, value: int): + self.name = name + self.value = value + + @self.app.mcp_tool() + def test_func() -> MyData: + """Test function""" + return MyData("test", 42) + + wrapper = test_func._function._func + context = json.dumps({"arguments": {}}) + result = asyncio.run(wrapper(context)) + result_obj = json.loads(result) + + self.assertIn("structuredContent", result_obj) + self.assertIsNotNone(result_obj["structuredContent"]) + + # Verify structured content contains the data + structured_obj = json.loads(result_obj["structuredContent"]) + self.assertEqual(structured_obj["name"], "test") + self.assertEqual(structured_obj["value"], 42)