Skip to content

Commit 39b413e

Browse files
shahar4499shahar4499
committed
add more schema info
1 parent 305352a commit 39b413e

File tree

1 file changed

+320
-4
lines changed

1 file changed

+320
-4
lines changed

‎fastapi_mcp/http_tools.py

Lines changed: 320 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,89 @@
1818
logger = logging.getLogger("fastapi_mcp")
1919

2020

21+
def resolve_schema_references(schema: Dict[str, Any], openapi_schema: Dict[str, Any]) -> Dict[str, Any]:
22+
"""
23+
Resolve schema references in OpenAPI schemas.
24+
25+
Args:
26+
schema: The schema that may contain references
27+
openapi_schema: The full OpenAPI schema to resolve references from
28+
29+
Returns:
30+
The schema with references resolved
31+
"""
32+
# Make a copy to avoid modifying the input schema
33+
schema = schema.copy()
34+
35+
# Handle $ref directly in the schema
36+
if "$ref" in schema:
37+
ref_path = schema["$ref"]
38+
# Standard OpenAPI references are in the format "#/components/schemas/ModelName"
39+
if ref_path.startswith("#/components/schemas/"):
40+
model_name = ref_path.split("/")[-1]
41+
if "components" in openapi_schema and "schemas" in openapi_schema["components"]:
42+
if model_name in openapi_schema["components"]["schemas"]:
43+
# Replace with the resolved schema
44+
ref_schema = openapi_schema["components"]["schemas"][model_name].copy()
45+
# Remove the $ref key and merge with the original schema
46+
schema.pop("$ref")
47+
schema.update(ref_schema)
48+
49+
# Handle array items
50+
if "type" in schema and schema["type"] == "array" and "items" in schema:
51+
schema["items"] = resolve_schema_references(schema["items"], openapi_schema)
52+
53+
# Handle object properties
54+
if "properties" in schema:
55+
for prop_name, prop_schema in schema["properties"].items():
56+
schema["properties"][prop_name] = resolve_schema_references(prop_schema, openapi_schema)
57+
58+
return schema
59+
60+
61+
def clean_schema_for_display(schema: Dict[str, Any]) -> Dict[str, Any]:
62+
"""
63+
Clean up a schema for display by removing internal fields.
64+
65+
Args:
66+
schema: The schema to clean
67+
68+
Returns:
69+
The cleaned schema
70+
"""
71+
# Make a copy to avoid modifying the input schema
72+
schema = schema.copy()
73+
74+
# Remove common internal fields that are not helpful for LLMs
75+
fields_to_remove = [
76+
"allOf",
77+
"anyOf",
78+
"oneOf",
79+
"nullable",
80+
"discriminator",
81+
"readOnly",
82+
"writeOnly",
83+
"xml",
84+
"externalDocs",
85+
]
86+
for field in fields_to_remove:
87+
if field in schema:
88+
schema.pop(field)
89+
90+
# Process nested properties
91+
if "properties" in schema:
92+
for prop_name, prop_schema in schema["properties"].items():
93+
if isinstance(prop_schema, dict):
94+
schema["properties"][prop_name] = clean_schema_for_display(prop_schema)
95+
96+
# Process array items
97+
if "type" in schema and schema["type"] == "array" and "items" in schema:
98+
if isinstance(schema["items"], dict):
99+
schema["items"] = clean_schema_for_display(schema["items"])
100+
101+
return schema
102+
103+
21104
def create_mcp_tools_from_openapi(app: FastAPI, mcp_server: FastMCP, base_url: str = None) -> None:
22105
"""
23106
Create MCP tools from a FastAPI app's OpenAPI schema.
@@ -77,6 +160,7 @@ def create_mcp_tools_from_openapi(app: FastAPI, mcp_server: FastMCP, base_url: s
77160
parameters=operation.get("parameters", []),
78161
request_body=operation.get("requestBody", {}),
79162
responses=operation.get("responses", {}),
163+
openapi_schema=openapi_schema,
80164
)
81165

82166

@@ -91,6 +175,7 @@ def create_http_tool(
91175
parameters: List[Dict[str, Any]],
92176
request_body: Dict[str, Any],
93177
responses: Dict[str, Any],
178+
openapi_schema: Dict[str, Any],
94179
) -> None:
95180
"""
96181
Create an MCP tool that makes an HTTP request to a FastAPI endpoint.
@@ -106,6 +191,7 @@ def create_http_tool(
106191
parameters: OpenAPI parameters
107192
request_body: OpenAPI request body
108193
responses: OpenAPI responses
194+
openapi_schema: The full OpenAPI schema
109195
"""
110196
# Build tool description
111197
tool_description = f"{summary}" if summary else f"{method.upper()} {path}"
@@ -115,21 +201,138 @@ def create_http_tool(
115201
# Add response schema information to description
116202
if responses:
117203
response_info = "\n\n### Responses:\n"
204+
205+
# Find the success response (usually 200 or 201)
206+
success_codes = ["200", "201", "202", 200, 201, 202]
207+
success_response = None
208+
for status_code in success_codes:
209+
if str(status_code) in responses:
210+
success_response = responses[str(status_code)]
211+
break
212+
213+
# Process all responses
118214
for status_code, response_data in responses.items():
119215
response_desc = response_data.get("description", "")
120216
response_info += f"\n**{status_code}**: {response_desc}"
121217

218+
# Highlight if this is the main success response
219+
if response_data == success_response:
220+
response_info += " (Success Response)"
221+
122222
# Add schema information if available
123223
if "content" in response_data:
124224
for content_type, content_data in response_data["content"].items():
125225
if "schema" in content_data:
126226
schema = content_data["schema"]
127227
response_info += f"\nContent-Type: {content_type}"
128228

129-
# Format schema information
130-
if "properties" in schema:
131-
response_info += "\n\nSchema:\n```json\n"
132-
response_info += json.dumps(schema, indent=2)
229+
# Resolve any schema references
230+
resolved_schema = resolve_schema_references(schema, openapi_schema)
231+
232+
# Clean the schema for display
233+
display_schema = clean_schema_for_display(resolved_schema)
234+
235+
# Get model name if it's a referenced model
236+
model_name = None
237+
model_examples = None
238+
items_model_name = None
239+
if "$ref" in schema:
240+
ref_path = schema["$ref"]
241+
if ref_path.startswith("#/components/schemas/"):
242+
model_name = ref_path.split("/")[-1]
243+
response_info += f"\nModel: {model_name}"
244+
# Try to get examples from the model
245+
model_examples = extract_model_examples_from_components(model_name, openapi_schema)
246+
247+
# Check if this is an array of items
248+
if schema.get("type") == "array" and "items" in schema and "$ref" in schema["items"]:
249+
items_ref_path = schema["items"]["$ref"]
250+
if items_ref_path.startswith("#/components/schemas/"):
251+
items_model_name = items_ref_path.split("/")[-1]
252+
response_info += f"\nArray of: {items_model_name}"
253+
254+
# Create example response based on schema type
255+
example_response = None
256+
257+
# Check if we have examples from the model
258+
if model_examples and len(model_examples) > 0:
259+
example_response = model_examples[0] # Use first example
260+
# Otherwise, try to create an example from the response definitions
261+
elif "examples" in response_data:
262+
# Use examples directly from response definition
263+
for example_key, example_data in response_data["examples"].items():
264+
if "value" in example_data:
265+
example_response = example_data["value"]
266+
break
267+
# If content has examples
268+
elif "examples" in content_data:
269+
for example_key, example_data in content_data["examples"].items():
270+
if "value" in example_data:
271+
example_response = example_data["value"]
272+
break
273+
# If content has example
274+
elif "example" in content_data:
275+
example_response = content_data["example"]
276+
277+
# Special handling for array of items
278+
if (
279+
not example_response
280+
and display_schema.get("type") == "array"
281+
and items_model_name == "Item"
282+
):
283+
example_response = [
284+
{
285+
"id": 1,
286+
"name": "Hammer",
287+
"description": "A tool for hammering nails",
288+
"price": 9.99,
289+
"tags": ["tool", "hardware"],
290+
},
291+
{
292+
"id": 2,
293+
"name": "Screwdriver",
294+
"description": "A tool for driving screws",
295+
"price": 7.99,
296+
"tags": ["tool", "hardware"],
297+
},
298+
] # type: ignore
299+
300+
# If we have an example response, add it to the docs
301+
if example_response:
302+
response_info += "\n\n**Example Response:**\n```json\n"
303+
response_info += json.dumps(example_response, indent=2)
304+
response_info += "\n```"
305+
# Otherwise generate an example from the schema
306+
else:
307+
generated_example = generate_example_from_schema(display_schema, model_name)
308+
if generated_example:
309+
response_info += "\n\n**Example Response:**\n```json\n"
310+
response_info += json.dumps(generated_example, indent=2)
311+
response_info += "\n```"
312+
313+
# Format schema information based on its type
314+
if display_schema.get("type") == "array" and "items" in display_schema:
315+
items_schema = display_schema["items"]
316+
# Check if items reference a model
317+
items_model_name = None
318+
if "$ref" in schema.get("items", {}):
319+
items_ref_path = schema["items"]["$ref"]
320+
if items_ref_path.startswith("#/components/schemas/"):
321+
items_model_name = items_ref_path.split("/")[-1]
322+
response_info += f"\nArray of: {items_model_name}"
323+
324+
response_info += (
325+
"\n\n**Output Schema:** Array of items with the following structure:\n```json\n"
326+
)
327+
response_info += json.dumps(items_schema, indent=2)
328+
response_info += "\n```"
329+
elif "properties" in display_schema:
330+
response_info += "\n\n**Output Schema:**\n```json\n"
331+
response_info += json.dumps(display_schema, indent=2)
332+
response_info += "\n```"
333+
else:
334+
response_info += "\n\n**Output Schema:**\n```json\n"
335+
response_info += json.dumps(display_schema, indent=2)
133336
response_info += "\n```"
134337

135338
tool_description += response_info
@@ -279,3 +482,116 @@ async def http_tool_function(**kwargs):
279482

280483
# Update the tool's parameters to use our custom schema instead of the auto-generated one
281484
tool.parameters = input_schema
485+
486+
487+
def extract_model_examples_from_components(
488+
model_name: str, openapi_schema: Dict[str, Any]
489+
) -> Optional[List[Dict[str, Any]]]:
490+
"""
491+
Extract examples from a model definition in the OpenAPI components.
492+
493+
Args:
494+
model_name: The name of the model to extract examples from
495+
openapi_schema: The full OpenAPI schema
496+
497+
Returns:
498+
List of example dictionaries if found, None otherwise
499+
"""
500+
if "components" not in openapi_schema or "schemas" not in openapi_schema["components"]:
501+
return None
502+
503+
if model_name not in openapi_schema["components"]["schemas"]:
504+
return None
505+
506+
schema = openapi_schema["components"]["schemas"][model_name]
507+
508+
# Look for examples in the schema
509+
examples = None
510+
511+
# Check for examples field directly (OpenAPI 3.1.0+)
512+
if "examples" in schema:
513+
examples = schema["examples"]
514+
# Check for example field (older OpenAPI versions)
515+
elif "example" in schema:
516+
examples = [schema["example"]]
517+
518+
return examples
519+
520+
521+
def generate_example_from_schema(schema: Dict[str, Any], model_name: Optional[str] = None) -> Any:
522+
"""
523+
Generate a simple example response from a JSON schema.
524+
525+
Args:
526+
schema: The JSON schema to generate an example from
527+
model_name: Optional model name for special handling
528+
529+
Returns:
530+
An example object based on the schema
531+
"""
532+
if not schema or not isinstance(schema, dict):
533+
return None
534+
535+
# Special handling for known model types
536+
if model_name == "Item":
537+
# Create a realistic Item example since this is commonly used
538+
return {
539+
"id": 1,
540+
"name": "Hammer",
541+
"description": "A tool for hammering nails",
542+
"price": 9.99,
543+
"tags": ["tool", "hardware"],
544+
}
545+
elif model_name == "HTTPValidationError":
546+
# Create a realistic validation error example
547+
return {"detail": [{"loc": ["body", "name"], "msg": "field required", "type": "value_error.missing"}]}
548+
549+
# Handle different types
550+
schema_type = schema.get("type")
551+
552+
if schema_type == "object":
553+
result = {}
554+
if "properties" in schema:
555+
for prop_name, prop_schema in schema["properties"].items():
556+
# Generate an example for each property
557+
prop_example = generate_example_from_schema(prop_schema)
558+
if prop_example is not None:
559+
result[prop_name] = prop_example
560+
return result
561+
562+
elif schema_type == "array":
563+
if "items" in schema:
564+
# Generate a single example item
565+
item_example = generate_example_from_schema(schema["items"])
566+
if item_example is not None:
567+
return [item_example]
568+
return []
569+
570+
elif schema_type == "string":
571+
# Check if there's a format
572+
format_type = schema.get("format")
573+
if format_type == "date-time":
574+
return "2023-01-01T00:00:00Z"
575+
elif format_type == "date":
576+
return "2023-01-01"
577+
elif format_type == "email":
578+
return "[email protected]"
579+
elif format_type == "uri":
580+
return "https://example.com"
581+
# Use title or property name if available
582+
return schema.get("title", "string")
583+
584+
elif schema_type == "integer":
585+
return 1
586+
587+
elif schema_type == "number":
588+
return 1.0
589+
590+
elif schema_type == "boolean":
591+
return True
592+
593+
elif schema_type == "null":
594+
return None
595+
596+
# Default case
597+
return None

0 commit comments

Comments
 (0)