diff --git a/fastapi_mcp/server.py b/fastapi_mcp/server.py index bb75106..54e5601 100644 --- a/fastapi_mcp/server.py +++ b/fastapi_mcp/server.py @@ -75,6 +75,12 @@ def __init__( Optional[AuthConfig], Doc("Configuration for MCP authentication"), ] = None, + prefer_structured_content: Annotated[ + bool, + Doc( + "Whether to prefer structured content over text content in responses. Defaults to False for backwards compatibility." + ), + ] = False, headers: Annotated[ List[str], Doc( @@ -108,6 +114,7 @@ def __init__( self._include_tags = include_tags self._exclude_tags = exclude_tags self._auth_config = auth_config + self._prefer_structured_content = prefer_structured_content if self._auth_config: self._auth_config = self._auth_config.model_validate(self._auth_config) @@ -150,7 +157,7 @@ async def handle_list_tools() -> List[types.Tool]: @mcp_server.call_tool() async def handle_call_tool( name: str, arguments: Dict[str, Any] - ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: + ) -> Union[List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]], Dict[str, Any]]: # Extract HTTP request info from MCP context http_request_info = None try: @@ -491,7 +498,7 @@ async def _execute_api_tool( Optional[HTTPRequestInfo], Doc("HTTP request info to forward to the actual API call"), ] = None, - ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: + ) -> Union[List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]], Dict[str, Any]]: """ Execute an MCP tool by making an HTTP request to the corresponding API endpoint. @@ -544,6 +551,7 @@ async def _execute_api_tool( response = await self._request(client, method, path, query, headers, body) # TODO: Better typing for the AsyncClientProtocol. It should return a ResponseProtocol that has a json() method that returns a dict/list/etc. + result = None try: result = response.json() result_text = json.dumps(result, indent=2, ensure_ascii=False) @@ -560,6 +568,9 @@ async def _execute_api_tool( f"Error calling {tool_name}. Status code: {response.status_code}. Response: {response.text}" ) + if result is not None and self._prefer_structured_content: + return result + try: return [types.TextContent(type="text", text=result_text)] except ValueError: diff --git a/tests/test_mcp_execute_api_tool.py b/tests/test_mcp_execute_api_tool.py index cc05d34..220ed30 100644 --- a/tests/test_mcp_execute_api_tool.py +++ b/tests/test_mcp_execute_api_tool.py @@ -1,4 +1,5 @@ import pytest +import json from unittest.mock import AsyncMock, patch, MagicMock from fastapi import FastAPI @@ -10,183 +11,291 @@ async def test_execute_api_tool_success(simple_fastapi_app: FastAPI): """Test successful execution of an API tool.""" mcp = FastApiMCP(simple_fastapi_app) - + # Mock the HTTP client response mock_response = MagicMock() mock_response.json.return_value = {"id": 1, "name": "Test Item"} mock_response.status_code = 200 mock_response.text = '{"id": 1, "name": "Test Item"}' - + # Mock the HTTP client mock_client = AsyncMock() mock_client.get.return_value = mock_response - + # Test parameters tool_name = "get_item" arguments = {"item_id": 1} - + # Execute the tool - with patch.object(mcp, '_http_client', mock_client): + with patch.object(mcp, "_http_client", mock_client): result = await mcp._execute_api_tool( - client=mock_client, - tool_name=tool_name, - arguments=arguments, - operation_map=mcp.operation_map + client=mock_client, tool_name=tool_name, arguments=arguments, operation_map=mcp.operation_map ) - + # Verify the result assert len(result) == 1 assert isinstance(result[0], TextContent) assert result[0].text == '{\n "id": 1,\n "name": "Test Item"\n}' - + # Verify the HTTP client was called correctly - mock_client.get.assert_called_once_with( - "/items/1", - params={}, - headers={} - ) + mock_client.get.assert_called_once_with("/items/1", params={}, headers={}) + + +@pytest.mark.asyncio +async def test_execute_api_tool_with_structured_content(simple_fastapi_app: FastAPI): + """Test execution of an API tool with structured content enabled.""" + mcp = FastApiMCP(simple_fastapi_app, prefer_structured_content=True) + + # Mock the HTTP client response + mock_response = MagicMock() + mock_response.json.return_value = {"id": 1, "name": "Test Item", "price": 10.0} + mock_response.status_code = 200 + mock_response.text = '{"id": 1, "name": "Test Item", "price": 10.0}' + + # Mock the HTTP client + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + + # Test parameters + tool_name = "get_item" + arguments = {"item_id": 1} + + # Execute the tool + with patch.object(mcp, "_http_client", mock_client): + result = await mcp._execute_api_tool( + client=mock_client, tool_name=tool_name, arguments=arguments, operation_map=mcp.operation_map + ) + + # Verify the result is structured content (dict) + assert isinstance(result, dict) + assert result["id"] == 1 + assert result["name"] == "Test Item" + assert result["price"] == 10.0 + + # Verify the HTTP client was called correctly + mock_client.get.assert_called_once_with("/items/1", params={}, headers={}) + + +@pytest.mark.asyncio +async def test_execute_api_tool_with_structured_content_list(simple_fastapi_app: FastAPI): + """Test execution of an API tool returning a list with structured content enabled.""" + mcp = FastApiMCP(simple_fastapi_app, prefer_structured_content=True) + + # Mock the HTTP client response + mock_response = MagicMock() + mock_response.json.return_value = [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}] + mock_response.status_code = 200 + mock_response.text = '[{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]' + + # Mock the HTTP client + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + + # Test parameters + tool_name = "list_items" + arguments = {"skip": 0, "limit": 2} + + # Execute the tool + with patch.object(mcp, "_http_client", mock_client): + result = await mcp._execute_api_tool( + client=mock_client, tool_name=tool_name, arguments=arguments, operation_map=mcp.operation_map + ) + + # Verify the result is structured content (list) + assert isinstance(result, list) + assert len(result) == 2 + assert result[0]["id"] == 1 + assert result[0]["name"] == "Item 1" + assert result[1]["id"] == 2 + assert result[1]["name"] == "Item 2" + + # Verify the HTTP client was called with query parameters + mock_client.get.assert_called_once_with("/items/", params={"skip": 0, "limit": 2}, headers={}) + + +@pytest.mark.asyncio +async def test_execute_api_tool_with_structured_content_disabled(simple_fastapi_app: FastAPI): + """Test execution of an API tool with structured content disabled (default behavior).""" + mcp = FastApiMCP(simple_fastapi_app, prefer_structured_content=False) + + # Mock the HTTP client response + mock_response = MagicMock() + mock_response.json.return_value = {"id": 1, "name": "Test Item"} + mock_response.status_code = 200 + mock_response.text = '{"id": 1, "name": "Test Item"}' + + # Mock the HTTP client + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + + # Test parameters + tool_name = "get_item" + arguments = {"item_id": 1} + + # Execute the tool + with patch.object(mcp, "_http_client", mock_client): + result = await mcp._execute_api_tool( + client=mock_client, tool_name=tool_name, arguments=arguments, operation_map=mcp.operation_map + ) + + # Verify the result is text content (default behavior) + assert len(result) == 1 + assert isinstance(result[0], TextContent) + assert result[0].text == '{\n "id": 1,\n "name": "Test Item"\n}' + + # Verify the HTTP client was called correctly + mock_client.get.assert_called_once_with("/items/1", params={}, headers={}) + + +@pytest.mark.asyncio +async def test_execute_api_tool_with_non_json_response_structured_content(simple_fastapi_app: FastAPI): + """Test execution of an API tool with non-JSON response when structured content is enabled.""" + mcp = FastApiMCP(simple_fastapi_app, prefer_structured_content=True) + + # Mock the HTTP client response with non-JSON content + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = "Plain text response" + mock_response.content = b"Plain text response" + mock_response.json.side_effect = json.JSONDecodeError("Expecting value", "", 0) + + # Mock the HTTP client + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + + # Test parameters + tool_name = "get_item" + arguments = {"item_id": 1} + + # Execute the tool + with patch.object(mcp, "_http_client", mock_client): + result = await mcp._execute_api_tool( + client=mock_client, tool_name=tool_name, arguments=arguments, operation_map=mcp.operation_map + ) + + # Verify the result is text content (fallback for non-JSON) + assert len(result) == 1 + assert isinstance(result[0], TextContent) + assert result[0].text == "Plain text response" + + # Verify the HTTP client was called correctly + mock_client.get.assert_called_once_with("/items/1", params={}, headers={}) @pytest.mark.asyncio async def test_execute_api_tool_with_query_params(simple_fastapi_app: FastAPI): """Test execution of an API tool with query parameters.""" mcp = FastApiMCP(simple_fastapi_app) - + # Mock the HTTP client response mock_response = MagicMock() mock_response.json.return_value = [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}] mock_response.status_code = 200 mock_response.text = '[{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]' - + # Mock the HTTP client mock_client = AsyncMock() mock_client.get.return_value = mock_response - + # Test parameters tool_name = "list_items" arguments = {"skip": 0, "limit": 2} - + # Execute the tool - with patch.object(mcp, '_http_client', mock_client): + with patch.object(mcp, "_http_client", mock_client): result = await mcp._execute_api_tool( - client=mock_client, - tool_name=tool_name, - arguments=arguments, - operation_map=mcp.operation_map + client=mock_client, tool_name=tool_name, arguments=arguments, operation_map=mcp.operation_map ) - + # Verify the result assert len(result) == 1 assert isinstance(result[0], TextContent) - + # Verify the HTTP client was called with query parameters - mock_client.get.assert_called_once_with( - "/items/", - params={"skip": 0, "limit": 2}, - headers={} - ) + mock_client.get.assert_called_once_with("/items/", params={"skip": 0, "limit": 2}, headers={}) @pytest.mark.asyncio async def test_execute_api_tool_with_body(simple_fastapi_app: FastAPI): """Test execution of an API tool with request body.""" mcp = FastApiMCP(simple_fastapi_app) - + # Mock the HTTP client response mock_response = MagicMock() mock_response.json.return_value = {"id": 1, "name": "New Item"} mock_response.status_code = 200 mock_response.text = '{"id": 1, "name": "New Item"}' - + # Mock the HTTP client mock_client = AsyncMock() mock_client.post.return_value = mock_response - + # Test parameters tool_name = "create_item" arguments = { - "item": { - "id": 1, - "name": "New Item", - "price": 10.0, - "tags": ["tag1"], - "description": "New item description" - } + "item": {"id": 1, "name": "New Item", "price": 10.0, "tags": ["tag1"], "description": "New item description"} } - + # Execute the tool - with patch.object(mcp, '_http_client', mock_client): + with patch.object(mcp, "_http_client", mock_client): result = await mcp._execute_api_tool( - client=mock_client, - tool_name=tool_name, - arguments=arguments, - operation_map=mcp.operation_map + client=mock_client, tool_name=tool_name, arguments=arguments, operation_map=mcp.operation_map ) - + # Verify the result assert len(result) == 1 assert isinstance(result[0], TextContent) - + # Verify the HTTP client was called with the request body - mock_client.post.assert_called_once_with( - "/items/", - params={}, - headers={}, - json=arguments - ) + mock_client.post.assert_called_once_with("/items/", params={}, headers={}, json=arguments) @pytest.mark.asyncio async def test_execute_api_tool_with_non_ascii_chars(simple_fastapi_app: FastAPI): """Test execution of an API tool with non-ASCII characters.""" mcp = FastApiMCP(simple_fastapi_app) - + # Test data with both ASCII and non-ASCII characters test_data = { "id": 1, "name": "你好 World", # Chinese characters + ASCII "price": 10.0, "tags": ["tag1", "标签2"], # Chinese characters in tags - "description": "这是一个测试描述" # All Chinese characters + "description": "这是一个测试描述", # All Chinese characters } - + # Mock the HTTP client response mock_response = MagicMock() mock_response.json.return_value = test_data mock_response.status_code = 200 - mock_response.text = '{"id": 1, "name": "你好 World", "price": 10.0, "tags": ["tag1", "标签2"], "description": "这是一个测试描述"}' - + mock_response.text = ( + '{"id": 1, "name": "你好 World", "price": 10.0, "tags": ["tag1", "标签2"], "description": "这是一个测试描述"}' + ) + # Mock the HTTP client mock_client = AsyncMock() mock_client.get.return_value = mock_response - + # Test parameters tool_name = "get_item" arguments = {"item_id": 1} - + # Execute the tool - with patch.object(mcp, '_http_client', mock_client): + with patch.object(mcp, "_http_client", mock_client): result = await mcp._execute_api_tool( - client=mock_client, - tool_name=tool_name, - arguments=arguments, - operation_map=mcp.operation_map + client=mock_client, tool_name=tool_name, arguments=arguments, operation_map=mcp.operation_map ) - + # Verify the result assert len(result) == 1 assert isinstance(result[0], TextContent) - + # Verify that the response contains both ASCII and non-ASCII characters response_text = result[0].text assert "你好" in response_text # Chinese characters preserved assert "World" in response_text # ASCII characters preserved assert "标签2" in response_text # Chinese characters in tags preserved assert "这是一个测试描述" in response_text # All Chinese description preserved - + # Verify the HTTP client was called correctly - mock_client.get.assert_called_once_with( - "/items/1", - params={}, - headers={} - ) + mock_client.get.assert_called_once_with("/items/1", params={}, headers={})