From 42f0c268121987fd62d053c1a3cfeee56b8e9843 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Tue, 17 Mar 2026 15:48:08 -0500 Subject: [PATCH 1/2] add support for structured content --- azure/functions/__init__.py | 14 +- azure/functions/decorators/function_app.py | 62 +++++++- azure/functions/decorators/mcp.py | 65 ++++++++ azure/functions/mcp.py | 88 +++++++++++ tests/decorators/test_mcp.py | 168 +++++++++++++++++++++ 5 files changed, 393 insertions(+), 4 deletions(-) diff --git a/azure/functions/__init__.py b/azure/functions/__init__.py index e3ffe6cf..70db47aa 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..fe374bc0 100644 --- a/azure/functions/decorators/function_app.py +++ b/azure/functions/decorators/function_app.py @@ -1604,7 +1604,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,6 +1615,7 @@ 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: @@ -1653,6 +1654,11 @@ def decorator(fb: FunctionBuilder) -> FunctionBuilder: # Wrap the original function @functools.wraps(target_func) async def wrapper(context: str, *args, **kwargs): + from azure.functions.mcp import ( + ContentBlock, CallToolResult) + from azure.functions.decorators.mcp import should_create_structured_content + import dataclasses + content = json.loads(context) arguments = content.get("arguments", {}) call_kwargs = {} @@ -1667,6 +1673,57 @@ async def wrapper(context: str, *args, **kwargs): result = target_func(**call_kwargs) if asyncio.iscoroutine(result): result = await result + + if result is None: + return str(result) + + # Handle CallToolResult - manual construction by user + if isinstance(result, CallToolResult): + result_dict = result.to_dict() + return json.dumps({ + "type": "call_tool_result", + "content": json.dumps(result_dict), + "structuredContent": json.dumps(result.structured_content) if result.structured_content else None + }) + + # 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) + }) + + # Handle single ContentBlock + if isinstance(result, ContentBlock): + block_dict = result.to_dict() + return json.dumps({ + "type": result.type, + "content": json.dumps(block_dict) + }) + + # Handle structured content generation when use_result_schema is True + if 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 json.dumps({ + "type": "text", + "content": json.dumps({"type": "text", "text": result_json}), + "structuredContent": result_json + }) + return str(result) wrapper.__signature__ = wrapper_sig @@ -1679,7 +1736,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=use_result_schema ) ) return fb diff --git a/azure/functions/decorators/mcp.py b/azure/functions/decorators/mcp.py index 435b8864..ae74aaa9 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) @@ -156,3 +161,63 @@ 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. + """ + if obj is None: + return False + 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 (unless explicitly marked) + + 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)): + 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..80374d84 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", 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..c81f7b19 100644 --- a/tests/decorators/test_mcp.py +++ b/tests/decorators/test_mcp.py @@ -504,3 +504,171 @@ 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.useResultSchema, 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") + 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") + 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") + self.assertEqual(block.uri, "file://logo.png") + + block_dict = block.to_dict() + self.assertEqual(block_dict["type"], "resource") + 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) + + From 874cac642fe26c2b74cf4fae35970f56e38f8373 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Wed, 18 Mar 2026 13:24:04 -0500 Subject: [PATCH 2/2] lint + tests --- azure/functions/decorators/function_app.py | 20 +- azure/functions/decorators/mcp.py | 18 +- azure/functions/mcp.py | 2 +- tests/decorators/test_mcp.py | 302 +++++++++++++++++++-- 4 files changed, 307 insertions(+), 35 deletions(-) diff --git a/azure/functions/decorators/function_app.py b/azure/functions/decorators/function_app.py index fe374bc0..c93b4fc8 100644 --- a/azure/functions/decorators/function_app.py +++ b/azure/functions/decorators/function_app.py @@ -1685,15 +1685,16 @@ async def wrapper(context: str, *args, **kwargs): "content": json.dumps(result_dict), "structuredContent": json.dumps(result.structured_content) if result.structured_content else None }) - + # Handle List[ContentBlock] - multiple content blocks - if isinstance(result, list) and all(isinstance(item, ContentBlock) for item in result): + 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) }) - + # Handle single ContentBlock if isinstance(result, ContentBlock): block_dict = result.to_dict() @@ -1701,7 +1702,7 @@ async def wrapper(context: str, *args, **kwargs): "type": result.type, "content": json.dumps(block_dict) }) - + # Handle structured content generation when use_result_schema is True if use_result_schema: # Check if we should create structured content @@ -1709,21 +1710,24 @@ async def wrapper(context: str, *args, **kwargs): # Serialize result as JSON for structured content # Handle dataclasses properly if dataclasses.is_dataclass(result): - result_json = json.dumps(dataclasses.asdict(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 - + result_json = json.dumps( + result) if not isinstance( + result, str) else result + # Return McpToolResult format with both text and structured content return json.dumps({ "type": "text", "content": json.dumps({"type": "text", "text": result_json}), "structuredContent": result_json }) - + return str(result) wrapper.__signature__ = wrapper_sig diff --git a/azure/functions/decorators/mcp.py b/azure/functions/decorators/mcp.py index ae74aaa9..8574a776 100644 --- a/azure/functions/decorators/mcp.py +++ b/azure/functions/decorators/mcp.py @@ -177,36 +177,36 @@ def has_mcp_content_marker(obj: typing.Any) -> bool: 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 (unless explicitly marked) - + 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)): 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 + 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: @@ -214,7 +214,7 @@ 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"]) diff --git a/azure/functions/mcp.py b/azure/functions/mcp.py index 80374d84..0d530d0b 100644 --- a/azure/functions/mcp.py +++ b/azure/functions/mcp.py @@ -77,7 +77,7 @@ class CallToolResult: """ Result type for MCP tool calls that allows manual construction of content blocks and structured content. - + Example: return CallToolResult( content=[ diff --git a/tests/decorators/test_mcp.py b/tests/decorators/test_mcp.py index c81f7b19..0a168918 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, }, ) @@ -518,12 +520,12 @@ def tearDown(self): 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__')) @@ -532,19 +534,19 @@ def __init__(self, value: str): 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)) @@ -554,11 +556,11 @@ def test_should_not_create_structured_content_for_primitives(self): 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)) @@ -568,22 +570,22 @@ def test_mcp_tool_with_use_result_schema_parameter(self): def test_tool(value: str): """Test tool with result schema""" return value - + trigger = test_tool._function._bindings[0] - self.assertEqual(trigger.useResultSchema, True) + 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__')) @@ -605,7 +607,7 @@ def test_image_content_block_creation(self): 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") @@ -622,7 +624,7 @@ def test_resource_link_block_creation(self): self.assertEqual(block.type, "resource") 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") self.assertEqual(block_dict["uri"], "https://example.com/resource") @@ -633,7 +635,7 @@ def test_resource_link_block_minimal(self): block = func.ResourceLinkBlock(uri="file://logo.png") self.assertEqual(block.type, "resource") self.assertEqual(block.uri, "file://logo.png") - + block_dict = block.to_dict() self.assertEqual(block_dict["type"], "resource") self.assertEqual(block_dict["uri"], "file://logo.png") @@ -649,12 +651,12 @@ def test_call_tool_result_creation(self): ], 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) @@ -665,10 +667,276 @@ def test_call_tool_result_without_structured_content(self): 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") + 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") + 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") + + # 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")