Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 64 additions & 10 deletions azure/functions/decorators/function_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -1663,16 +1663,44 @@ def decorator(fb: FunctionBuilder) -> FunctionBuilder:
args = typing.get_args(return_annotation)

# 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
if origin in (list, List):
# For List[T], check if T is an MCP type
try:
if len(args) > 0:
list_item_type = args[0]
# Check if it's a direct MCP type
if isinstance(list_item_type, type):
if hasattr(list_item_type, '__module__'):
module = list_item_type.__module__
if (module
and (module.startswith('mcp.types')
or module == 'mcp.types')):
is_mcp_sdk_type = True
# Check if it's a Union of MCP types
elif hasattr(list_item_type, '__origin__'):
union_origin = typing.get_origin(
list_item_type)
if union_origin is Union:
union_args = typing.get_args(
list_item_type)
for union_arg in union_args:
if (isinstance(union_arg, type)
and union_arg is not
type(None)):
if hasattr(union_arg,
'__module__'):
module = (
union_arg.__module__)
if (module
and (module.startswith(
'mcp.types')
or module
== 'mcp.types')):
is_mcp_sdk_type = True
break
except (ImportError, TypeError, AttributeError,
IndexError):
pass

# Check for Optional[T] where T is an MCP type
if origin is Union:
Expand Down Expand Up @@ -1801,6 +1829,32 @@ async def wrapper(context: str, *args, **kwargs):
"structuredContent": structured_content_json
}))

# Handle lists of MCP SDK content blocks
# Wrap them in a CallToolResult structure
elif isinstance(result, list) and len(result) > 0:
first_item = result[0]
if _is_mcp_sdk_type(first_item):
# Serialize all blocks in the list
from ..mcp import _serialize_content_block
blocks_list = [_serialize_content_block(block)
for block in result]

# Create a CallToolResult-like structure
# containing the blocks
call_tool_result = {
"content": blocks_list
}
full_result_json = json.dumps(call_tool_result)

# Return in CallToolResult format
# (list of blocks doesn't have separate
# structuredContent)
return str(json.dumps({
"type": "call_tool_result",
"content": full_result_json,
"structuredContent": None
}))

# Handle all other MCP SDK types
# (TextContent, ImageContent, ResourceLink, etc.)
elif _is_mcp_sdk_type(result):
Expand Down
40 changes: 40 additions & 0 deletions tests/decorators/test_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -675,6 +675,18 @@ def get_texts() -> List[TextContent]:
trigger = get_texts._function._bindings[0]
self.assertTrue(trigger.use_result_schema)

def test_auto_detect_list_union_mcp_types(self):
"""Test auto-detection of List[Union[MCP types]] return type"""
from typing import Union

@self.app.mcp_tool()
def get_mixed_content() -> List[Union[TextContent, ImageContent]]:
"""Returns mixed content blocks"""
return [TextContent(type="text", text="test")]

trigger = get_mixed_content._function._bindings[0]
self.assertTrue(trigger.use_result_schema)

def test_auto_detect_optional_mcp_image_content(self):
"""Test auto-detection of Optional[ImageContent] return type"""
@self.app.mcp_tool()
Expand Down Expand Up @@ -907,6 +919,34 @@ def test_func() -> str:
self.assertIn("content", result_obj)
self.assertIn("structuredContent", result_obj)

def test_structured_content_in_list_of_mcp_types(self):
"""Test that List[MCP SDK types] includes structuredContent"""
@self.app.mcp_tool()
def test_func() -> List[TextContent]:
"""Test function"""
return [
TextContent(type="text", text="First item"),
TextContent(type="text", text="Second item")
]

wrapper = test_func._function._func
context = json.dumps({"arguments": {}})
result = asyncio.run(wrapper(context))
result_obj = json.loads(result)

# List of content blocks is wrapped as CallToolResult
self.assertIn("type", result_obj)
self.assertEqual(result_obj["type"], "call_tool_result")
self.assertIn("content", result_obj)
self.assertIn("structuredContent", result_obj)

# Content contains the CallToolResult structure with the blocks
content_obj = json.loads(result_obj["content"])
self.assertIn("content", content_obj)
self.assertEqual(len(content_obj["content"]), 2)
self.assertEqual(content_obj["content"][0]["text"], "First item")
self.assertEqual(content_obj["content"][1]["text"], "Second item")


class TestMCPPackageNotInstalled(unittest.TestCase):
"""Tests for graceful degradation when mcp package is not installed"""
Expand Down
Loading