mirror of
https://github.com/tadata-org/fastapi_mcp.git
synced 2025-04-13 23:32:11 +03:00
remove dead code
This commit is contained in:
Shahar Abramov
@@ -4,96 +4,7 @@ OpenAPI utility functions for FastAPI-MCP.
|
||||
This module provides utility functions for working with OpenAPI schemas.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from datetime import date, datetime
|
||||
from decimal import Decimal
|
||||
from uuid import UUID
|
||||
|
||||
PYTHON_TYPE_IMPORTS = {
|
||||
"List": List,
|
||||
"Dict": Dict,
|
||||
"Any": Any,
|
||||
"Optional": Optional,
|
||||
"Union": Union,
|
||||
"date": date,
|
||||
"datetime": datetime,
|
||||
"Decimal": Decimal,
|
||||
"UUID": UUID,
|
||||
}
|
||||
# Type mapping from OpenAPI types to Python types
|
||||
OPENAPI_PYTHON_TYPES_MAP = {
|
||||
# Core data types (OpenAPI 3.x)
|
||||
"string": "str",
|
||||
"number": "Union[float, Decimal]", # Could be float or Decimal for precision
|
||||
"integer": "int",
|
||||
"boolean": "bool",
|
||||
"null": "None",
|
||||
# Complex types
|
||||
"object": "Dict[str, Any]", # More specific than Dict[Any, Any]
|
||||
"array": "List[Any]",
|
||||
# Numeric formats
|
||||
"int32": "int",
|
||||
"int64": "int",
|
||||
"float": "float",
|
||||
"double": "float",
|
||||
"decimal": "Decimal",
|
||||
# String formats - Common
|
||||
"date": "date", # datetime.date
|
||||
"date-time": "datetime", # datetime.datetime
|
||||
"time": "str", # Could use datetime.time
|
||||
"duration": "str", # Could use datetime.timedelta
|
||||
"password": "str",
|
||||
"byte": "bytes", # base64 encoded
|
||||
"binary": "bytes", # raw binary
|
||||
# String formats - Extended
|
||||
"email": "str",
|
||||
"uuid": "UUID", # uuid.UUID
|
||||
"uri": "str",
|
||||
"uri-reference": "str",
|
||||
"uri-template": "str",
|
||||
"url": "str",
|
||||
"hostname": "str",
|
||||
"ipv4": "str",
|
||||
"ipv6": "str",
|
||||
"regex": "str",
|
||||
"json-pointer": "str",
|
||||
"relative-json-pointer": "str",
|
||||
# Rich text formats
|
||||
"markdown": "str",
|
||||
"html": "str",
|
||||
# Media types
|
||||
"image/*": "bytes",
|
||||
"audio/*": "bytes",
|
||||
"video/*": "bytes",
|
||||
"application/*": "bytes",
|
||||
# Special formats
|
||||
"format": "str", # Custom format string
|
||||
"pattern": "str", # Regular expression pattern
|
||||
"contentEncoding": "str", # e.g., base64, quoted-printable
|
||||
"contentMediaType": "str", # MIME type
|
||||
# Additional numeric formats
|
||||
"currency": "Decimal", # For precise decimal calculations
|
||||
"percentage": "float",
|
||||
# Geographic coordinates
|
||||
"latitude": "float",
|
||||
"longitude": "float",
|
||||
# Time-related
|
||||
"timezone": "str", # Could use zoneinfo.ZoneInfo in Python 3.9+
|
||||
"unix-time": "int", # Unix timestamp
|
||||
"iso-week-date": "str", # ISO 8601 week date
|
||||
# Specialized string formats
|
||||
"isbn": "str",
|
||||
"issn": "str",
|
||||
"iban": "str",
|
||||
"credit-card": "str",
|
||||
"phone": "str",
|
||||
"postal-code": "str",
|
||||
"language-code": "str", # ISO 639 language codes
|
||||
"country-code": "str", # ISO 3166 country codes
|
||||
"currency-code": "str", # ISO 4217 currency codes
|
||||
# Default fallback
|
||||
"unknown": "Any",
|
||||
}
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
def get_single_param_type_from_schema(param_schema: Dict[str, Any]) -> str:
|
||||
@@ -111,21 +22,6 @@ def get_single_param_type_from_schema(param_schema: Dict[str, Any]) -> str:
|
||||
return param_schema.get("type", "string")
|
||||
|
||||
|
||||
def get_python_type_and_default(parsed_param_schema: Dict[str, Any]) -> tuple[str, bool]:
|
||||
"""
|
||||
Parse parameters into a python type and default value string.
|
||||
Returns:
|
||||
A tuple containing:
|
||||
- A string representing the Python type annotation (e.g. "str", "int = 0", etc.)
|
||||
- A boolean indicating whether a default value is present
|
||||
"""
|
||||
# Handle direct type specification
|
||||
python_type = OPENAPI_PYTHON_TYPES_MAP.get(parsed_param_schema.get("type", ""), "Any")
|
||||
if "default" in parsed_param_schema:
|
||||
return f"{python_type} = {parsed_param_schema.get('default')}", True
|
||||
return python_type, False
|
||||
|
||||
|
||||
def resolve_schema_references(schema_part: Dict[str, Any], reference_schema: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Resolve schema references in OpenAPI schemas.
|
||||
|
||||
@@ -13,8 +13,8 @@ from mcp.server.lowlevel.server import Server
|
||||
from mcp.server.sse import SseServerTransport
|
||||
import mcp.types as types
|
||||
|
||||
from .openapi.convert import convert_openapi_to_mcp_tools
|
||||
from .execute import execute_api_tool
|
||||
from fastapi_mcp.openapi.convert import convert_openapi_to_mcp_tools
|
||||
from fastapi_mcp.execute import execute_api_tool
|
||||
|
||||
|
||||
def create_mcp_server(
|
||||
|
||||
@@ -1,69 +0,0 @@
|
||||
"""
|
||||
Tests for the fastapi_mcp server module.
|
||||
This tests the creation and mounting of MCP servers to FastAPI applications.
|
||||
"""
|
||||
|
||||
from fastapi import FastAPI
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
from fastapi_mcp import create_mcp_server, add_mcp_server
|
||||
|
||||
|
||||
def test_create_mcp_server():
|
||||
"""Test creating an MCP server from a FastAPI app."""
|
||||
app = FastAPI(title="Test App", description="Test Description")
|
||||
|
||||
# Test with default parameters
|
||||
mcp_server = create_mcp_server(app)
|
||||
assert isinstance(mcp_server, FastMCP), "Should return a FastMCP instance"
|
||||
assert mcp_server.name == "Test App", "Server name should match app title"
|
||||
assert mcp_server.instructions == "Test Description", "Server description should match app description"
|
||||
|
||||
# Test with custom parameters
|
||||
custom_mcp_server = create_mcp_server(app, name="Custom Name", description="Custom Description")
|
||||
assert custom_mcp_server.name == "Custom Name", "Server name should match provided name"
|
||||
assert custom_mcp_server.instructions == "Custom Description", (
|
||||
"Server description should match provided description"
|
||||
)
|
||||
|
||||
|
||||
def test_server_configuration():
|
||||
"""Test that server configuration options are properly set."""
|
||||
app = FastAPI(title="Test API")
|
||||
|
||||
# Test default configuration
|
||||
mcp_server = create_mcp_server(app)
|
||||
assert mcp_server._tool_manager is not None, "Tool manager should be created"
|
||||
assert mcp_server._resource_manager is not None, "Resource manager should be created"
|
||||
assert mcp_server._prompt_manager is not None, "Prompt manager should be created"
|
||||
|
||||
# Test custom tool registration
|
||||
@mcp_server.tool()
|
||||
async def test_tool():
|
||||
"""Test tool"""
|
||||
return "Test result"
|
||||
|
||||
tools = mcp_server._tool_manager.list_tools()
|
||||
test_tool = next((t for t in tools if t.name == "test_tool"), None) # noqa: F811
|
||||
assert test_tool is not None, "Custom tool should be registered"
|
||||
assert test_tool.description == "Test tool", "Tool description should be preserved"
|
||||
assert test_tool.is_async is True, "Async tools should be detected correctly"
|
||||
|
||||
|
||||
def test_add_mcp_server_components():
|
||||
"""Test that add_mcp_server correctly adds all components."""
|
||||
app = FastAPI()
|
||||
|
||||
# Test with default parameters
|
||||
mcp_server = add_mcp_server(app, serve_tools=False) # Don't actually serve tools to avoid server setup
|
||||
assert isinstance(mcp_server, FastMCP), "Should return a FastMCP instance"
|
||||
assert mcp_server._mcp_server is not None, "MCP server should be created"
|
||||
|
||||
# Test custom tool addition
|
||||
@mcp_server.tool()
|
||||
async def test_tool():
|
||||
"""Test tool"""
|
||||
return "Test result"
|
||||
|
||||
tools = [t.name for t in mcp_server._tool_manager.list_tools()]
|
||||
assert "test_tool" in tools, "Custom tool should be registered"
|
||||
Reference in New Issue
Block a user