Skip to content

Commit 5577ff7

Browse files
bhosmer-antbhosmer-ant
committed
update README.md and add example
1 parent fff4cf9 commit 5577ff7

File tree

2 files changed

+300
-0
lines changed

2 files changed

+300
-0
lines changed

‎README.md‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -762,6 +762,61 @@ The lifespan API provides:
762762
- Access to initialized resources through the request context in handlers
763763
- Type-safe context passing between lifespan and request handlers
764764

765+
#### Structured Output Support
766+
767+
The low-level server supports structured output for tools, allowing you to return both human-readable content and machine-readable structured data. Tools can define an `outputSchema` to validate their structured output:
768+
769+
```python
770+
@server.list_tools()
771+
async def list_tools() -> list[types.Tool]:
772+
return [
773+
types.Tool(
774+
name="calculate",
775+
description="Perform mathematical calculations",
776+
inputSchema={
777+
"type": "object",
778+
"properties": {
779+
"expression": {"type": "string", "description": "Math expression"}
780+
},
781+
"required": ["expression"]
782+
},
783+
outputSchema={
784+
"type": "object",
785+
"properties": {
786+
"result": {"type": "number"},
787+
"expression": {"type": "string"}
788+
},
789+
"required": ["result", "expression"]
790+
}
791+
)
792+
]
793+
794+
@server.call_tool()
795+
async def call_tool(name: str, arguments: dict) -> tuple[list[types.TextContent], dict]:
796+
if name == "calculate":
797+
expression = arguments["expression"]
798+
try:
799+
result = eval(expression) # Note: Use a safe math parser in production
800+
801+
# Return both human-readable content and structured data
802+
content = [types.TextContent(
803+
type="text",
804+
text=f"The result of {expression} is {result}"
805+
)]
806+
structured = {"result": result, "expression": expression}
807+
808+
return (content, structured)
809+
except Exception as e:
810+
raise ValueError(f"Calculation error: {str(e)}")
811+
```
812+
813+
Tools can return data in three ways:
814+
1. **Content only**: Return a list of content blocks (default behavior)
815+
2. **Structured data only**: Return a dictionary that will be serialized to JSON
816+
3. **Both**: Return a tuple of (content, structured_data)
817+
818+
When an `outputSchema` is defined, the server automatically validates the structured output against the schema. This ensures type safety and helps catch errors early.
819+
765820
```python
766821
import mcp.server.stdio
767822
import mcp.types as types

‎examples/servers/structured_output_lowlevel.py‎

Lines changed: 245 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,245 @@
1+
#!/usr/bin/env python3
2+
"""
3+
Example low-level MCP server demonstrating structured output support.
4+
5+
This example shows how to use the low-level server API to return both
6+
human-readable content and machine-readable structured data from tools,
7+
with automatic validation against output schemas.
8+
9+
The low-level API provides direct control over request handling and
10+
allows tools to return different types of responses:
11+
1. Content only (list of content blocks)
12+
2. Structured data only (dict that gets serialized to JSON)
13+
3. Both content and structured data (tuple)
14+
"""
15+
16+
import asyncio
17+
from datetime import datetime
18+
from typing import Any
19+
20+
import mcp.server.stdio
21+
import mcp.types as types
22+
from mcp.server.lowlevel import NotificationOptions, Server
23+
from mcp.server.models import InitializationOptions
24+
25+
26+
# Create low-level server instance
27+
server = Server("structured-output-lowlevel-example")
28+
29+
30+
@server.list_tools()
31+
async def list_tools() -> list[types.Tool]:
32+
"""List available tools with their schemas."""
33+
return [
34+
types.Tool(
35+
name="analyze_text",
36+
description="Analyze text and return structured insights",
37+
inputSchema={
38+
"type": "object",
39+
"properties": {"text": {"type": "string", "description": "Text to analyze"}},
40+
"required": ["text"],
41+
},
42+
outputSchema={
43+
"type": "object",
44+
"properties": {
45+
"word_count": {"type": "integer"},
46+
"char_count": {"type": "integer"},
47+
"sentence_count": {"type": "integer"},
48+
"most_common_words": {
49+
"type": "array",
50+
"items": {
51+
"type": "object",
52+
"properties": {"word": {"type": "string"}, "count": {"type": "integer"}},
53+
"required": ["word", "count"],
54+
},
55+
},
56+
},
57+
"required": ["word_count", "char_count", "sentence_count", "most_common_words"],
58+
},
59+
),
60+
types.Tool(
61+
name="get_weather",
62+
description="Get weather information (simulated)",
63+
inputSchema={
64+
"type": "object",
65+
"properties": {"city": {"type": "string", "description": "City name"}},
66+
"required": ["city"],
67+
},
68+
outputSchema={
69+
"type": "object",
70+
"properties": {
71+
"temperature": {"type": "number"},
72+
"conditions": {"type": "string"},
73+
"humidity": {"type": "integer", "minimum": 0, "maximum": 100},
74+
"wind_speed": {"type": "number"},
75+
"timestamp": {"type": "string", "format": "date-time"},
76+
},
77+
"required": ["temperature", "conditions", "humidity", "wind_speed", "timestamp"],
78+
},
79+
),
80+
types.Tool(
81+
name="calculate_statistics",
82+
description="Calculate statistics for a list of numbers",
83+
inputSchema={
84+
"type": "object",
85+
"properties": {
86+
"numbers": {
87+
"type": "array",
88+
"items": {"type": "number"},
89+
"description": "List of numbers to analyze",
90+
}
91+
},
92+
"required": ["numbers"],
93+
},
94+
outputSchema={
95+
"type": "object",
96+
"properties": {
97+
"mean": {"type": "number"},
98+
"median": {"type": "number"},
99+
"min": {"type": "number"},
100+
"max": {"type": "number"},
101+
"sum": {"type": "number"},
102+
"count": {"type": "integer"},
103+
},
104+
"required": ["mean", "median", "min", "max", "sum", "count"],
105+
},
106+
),
107+
]
108+
109+
110+
@server.call_tool()
111+
async def call_tool(name: str, arguments: dict[str, Any]) -> Any:
112+
"""
113+
Handle tool calls with structured output.
114+
115+
This low-level handler demonstrates the three ways to return data:
116+
1. Return a list of content blocks (traditional approach)
117+
2. Return a dict (gets serialized to JSON and included as structuredContent)
118+
3. Return a tuple of (content, structured_data) for both
119+
"""
120+
121+
if name == "analyze_text":
122+
text = arguments["text"]
123+
124+
# Analyze the text
125+
words = text.split()
126+
word_count = len(words)
127+
char_count = len(text)
128+
sentences = text.replace("?", ".").replace("!", ".").split(".")
129+
sentence_count = len([s for s in sentences if s.strip()])
130+
131+
# Count word frequencies
132+
word_freq = {}
133+
for word in words:
134+
word_lower = word.lower().strip('.,!?;:"')
135+
if word_lower:
136+
word_freq[word_lower] = word_freq.get(word_lower, 0) + 1
137+
138+
# Get top 5 most common words
139+
most_common = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:5]
140+
most_common_words = [{"word": word, "count": count} for word, count in most_common]
141+
142+
# Example 3: Return both content and structured data
143+
# The low-level server will validate the structured data against outputSchema
144+
content = [
145+
types.TextContent(
146+
type="text",
147+
text=f"Text analysis complete:\n"
148+
f"- {word_count} words\n"
149+
f"- {char_count} characters\n"
150+
f"- {sentence_count} sentences\n"
151+
f"- Most common words: {', '.join(w['word'] for w in most_common_words)}",
152+
)
153+
]
154+
155+
structured = {
156+
"word_count": word_count,
157+
"char_count": char_count,
158+
"sentence_count": sentence_count,
159+
"most_common_words": most_common_words,
160+
}
161+
162+
return (content, structured)
163+
164+
elif name == "get_weather":
165+
city = arguments["city"]
166+
167+
# Simulate weather data (in production, call a real weather API)
168+
import random
169+
170+
weather_conditions = ["sunny", "cloudy", "rainy", "partly cloudy", "foggy"]
171+
172+
weather_data = {
173+
"temperature": round(random.uniform(0, 35), 1),
174+
"conditions": random.choice(weather_conditions),
175+
"humidity": random.randint(30, 90),
176+
"wind_speed": round(random.uniform(0, 30), 1),
177+
"timestamp": datetime.now().isoformat(),
178+
}
179+
180+
# Example 2: Return structured data only
181+
# The low-level server will serialize this to JSON content automatically
182+
return weather_data
183+
184+
elif name == "calculate_statistics":
185+
numbers = arguments["numbers"]
186+
187+
if not numbers:
188+
raise ValueError("Cannot calculate statistics for empty list")
189+
190+
sorted_nums = sorted(numbers)
191+
count = len(numbers)
192+
193+
# Calculate statistics
194+
mean = sum(numbers) / count
195+
196+
if count % 2 == 0:
197+
median = (sorted_nums[count // 2 - 1] + sorted_nums[count // 2]) / 2
198+
else:
199+
median = sorted_nums[count // 2]
200+
201+
stats = {
202+
"mean": mean,
203+
"median": median,
204+
"min": sorted_nums[0],
205+
"max": sorted_nums[-1],
206+
"sum": sum(numbers),
207+
"count": count,
208+
}
209+
210+
# Example 3: Return both content and structured data
211+
content = [
212+
types.TextContent(
213+
type="text",
214+
text=f"Statistics for {count} numbers:\n"
215+
f"Mean: {stats['mean']:.2f}, Median: {stats['median']:.2f}\n"
216+
f"Range: {stats['min']} to {stats['max']}\n"
217+
f"Sum: {stats['sum']}",
218+
)
219+
]
220+
221+
return (content, stats)
222+
223+
else:
224+
raise ValueError(f"Unknown tool: {name}")
225+
226+
227+
async def run():
228+
"""Run the low-level server using stdio transport."""
229+
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
230+
await server.run(
231+
read_stream,
232+
write_stream,
233+
InitializationOptions(
234+
server_name="structured-output-lowlevel-example",
235+
server_version="0.1.0",
236+
capabilities=server.get_capabilities(
237+
notification_options=NotificationOptions(),
238+
experimental_capabilities={},
239+
),
240+
),
241+
)
242+
243+
244+
if __name__ == "__main__":
245+
asyncio.run(run())

0 commit comments

Comments
 (0)