mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-10-21 04:11:25 +02:00
68 lines
1.9 KiB
Python
68 lines
1.9 KiB
Python
"""Hydration of ``Client.call_tool(...).data`` from the server's output schema.
|
|
|
|
The client rebuilds Python types from the output schema, so every ``format``
|
|
and positional item schema the server emits has to survive that round trip.
|
|
"""
|
|
|
|
from datetime import date, datetime, time, timedelta
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from fastmcp import Client, FastMCP
|
|
|
|
|
|
class Reading(BaseModel):
|
|
ts: datetime
|
|
day: date
|
|
at: time
|
|
dur: timedelta
|
|
ident: UUID
|
|
|
|
|
|
class TestStructuredResultHydration:
|
|
async def test_annotated_scalar_formats_arrive_as_python_types(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def read() -> Reading:
|
|
return Reading(
|
|
ts=datetime(2024, 1, 1, 12),
|
|
day=date(2024, 1, 1),
|
|
at=time(1, 2),
|
|
dur=timedelta(days=1),
|
|
ident=UUID(int=7),
|
|
)
|
|
|
|
async with Client(mcp) as client:
|
|
data = (await client.call_tool("read")).data
|
|
|
|
assert data.ts == datetime(2024, 1, 1, 12)
|
|
assert data.day == date(2024, 1, 1)
|
|
assert data.at == time(1, 2)
|
|
assert data.dur == timedelta(days=1)
|
|
assert data.ident == UUID(int=7)
|
|
|
|
async def test_fixed_length_tuple_result_keeps_positions(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def first_seen() -> tuple[date, int]:
|
|
return (date(2026, 8, 27), 3)
|
|
|
|
async with Client(mcp) as client:
|
|
data = (await client.call_tool("first_seen")).data
|
|
|
|
assert data == (date(2026, 8, 27), 3)
|
|
|
|
async def test_formats_hydrate_inside_collections(self):
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def recent() -> dict[str, list[date]]:
|
|
return {"days": [date(2026, 8, 27), date(2026, 8, 28)]}
|
|
|
|
async with Client(mcp) as client:
|
|
data = (await client.call_tool("recent")).data
|
|
|
|
assert data["days"] == [date(2026, 8, 27), date(2026, 8, 28)]
|