mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-26 15:34:18 +02:00
189 lines
6 KiB
Python
189 lines
6 KiB
Python
"""Tests for CallArgument and Depends bindings through FastMCP's resolution."""
|
|
|
|
import pytest
|
|
from mcp_types import TextContent
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.dependencies import CallArgument, CycleError, Depends
|
|
from fastmcp.server.dependencies import resolve_dependencies
|
|
|
|
|
|
@pytest.fixture
|
|
def mcp():
|
|
"""Create a FastMCP server for testing."""
|
|
return FastMCP("test-server")
|
|
|
|
|
|
async def test_bare_call_argument_reads_tool_parameter(mcp: FastMCP):
|
|
"""A bare CallArgument takes the value of the same-named tool parameter."""
|
|
|
|
def get_greeting(name: str = CallArgument()) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
@mcp.tool()
|
|
async def greet(name: str, greeting: str = Depends(get_greeting)) -> str:
|
|
return greeting
|
|
|
|
result = await mcp.call_tool("greet", {"name": "Alice"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "Hello, Alice!"
|
|
|
|
|
|
async def test_named_call_argument_reads_tool_parameter(mcp: FastMCP):
|
|
"""CallArgument("name") reads a tool parameter with a different name."""
|
|
|
|
def get_greeting(who: str = CallArgument("name")) -> str:
|
|
return f"Hello, {who}!"
|
|
|
|
@mcp.tool()
|
|
async def greet(name: str, greeting: str = Depends(get_greeting)) -> str:
|
|
return greeting
|
|
|
|
result = await mcp.call_tool("greet", {"name": "Bob"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "Hello, Bob!"
|
|
|
|
|
|
async def test_call_argument_in_binding(mcp: FastMCP):
|
|
"""A CallArgument binding wires a tool parameter to a factory parameter."""
|
|
|
|
def get_account(user_id: str) -> dict[str, str]:
|
|
return {"id": user_id, "plan": "pro"}
|
|
|
|
@mcp.tool()
|
|
async def show_account(
|
|
owner: str,
|
|
account: dict[str, str] = Depends(get_account, user_id=CallArgument("owner")),
|
|
) -> str:
|
|
return f"{account['id']}:{account['plan']}"
|
|
|
|
result = await mcp.call_tool("show_account", {"owner": "alice"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "alice:pro"
|
|
|
|
|
|
async def test_plain_value_binding(mcp: FastMCP):
|
|
"""A binding that is not a Dependency passes through to the factory as-is."""
|
|
|
|
def get_url(scheme: str) -> str:
|
|
return f"{scheme}://example.com"
|
|
|
|
@mcp.tool()
|
|
async def fetch(path: str, url: str = Depends(get_url, scheme="https")) -> str:
|
|
return f"{url}/{path}"
|
|
|
|
result = await mcp.call_tool("fetch", {"path": "docs"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "https://example.com/docs"
|
|
|
|
|
|
async def test_binding_replaces_factory_depends_default(mcp: FastMCP):
|
|
"""A binding replaces the factory's own Depends default, which never runs."""
|
|
|
|
default_calls = 0
|
|
|
|
def get_default_region() -> str:
|
|
nonlocal default_calls
|
|
default_calls += 1
|
|
return "us-east-1"
|
|
|
|
def get_bucket(region: str = Depends(get_default_region)) -> str:
|
|
return f"bucket-{region}"
|
|
|
|
@mcp.tool()
|
|
async def store(
|
|
data: str, bucket: str = Depends(get_bucket, region="eu-west-1")
|
|
) -> str:
|
|
return bucket
|
|
|
|
result = await mcp.call_tool("store", {"data": "payload"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "bucket-eu-west-1"
|
|
assert default_calls == 0
|
|
|
|
|
|
async def test_optional_call_argument_yields_none(mcp: FastMCP):
|
|
"""CallArgument(optional=True) yields None for a name the tool lacks."""
|
|
|
|
def get_note(tenant: str | None = CallArgument("tenant", optional=True)) -> str:
|
|
return f"tenant={tenant}"
|
|
|
|
@mcp.tool()
|
|
async def report(topic: str, note: str = Depends(get_note)) -> str:
|
|
return note
|
|
|
|
result = await mcp.call_tool("report", {"topic": "sales"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "tenant=None"
|
|
|
|
|
|
async def test_sibling_dependency_resolves_once(mcp: FastMCP):
|
|
"""A CallArgument reference to a dependency-backed sibling shares one value."""
|
|
|
|
session_calls = 0
|
|
|
|
def get_session() -> str:
|
|
nonlocal session_calls
|
|
session_calls += 1
|
|
return "session-1"
|
|
|
|
def audit(session: str = CallArgument()) -> str:
|
|
return f"audit:{session}"
|
|
|
|
@mcp.tool()
|
|
async def act(
|
|
step: str,
|
|
session: str = Depends(get_session),
|
|
log: str = Depends(audit),
|
|
) -> str:
|
|
return f"{log}|{session}"
|
|
|
|
result = await mcp.call_tool("act", {"step": "one"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "audit:session-1|session-1"
|
|
assert session_calls == 1
|
|
|
|
|
|
async def test_call_argument_cycle_raises_cycle_error():
|
|
"""CallArgument references that form a cycle raise CycleError with the path."""
|
|
|
|
def get_a(b: str = CallArgument()) -> str:
|
|
return b
|
|
|
|
def get_b(a: str = CallArgument()) -> str:
|
|
return a
|
|
|
|
async def entangled(a: str = Depends(get_a), b: str = Depends(get_b)) -> str:
|
|
return f"{a}{b}"
|
|
|
|
with pytest.raises(CycleError, match="a -> b -> a"):
|
|
async with resolve_dependencies(entangled, {}):
|
|
pass
|
|
|
|
|
|
async def test_colliding_argument_never_reaches_call_argument(mcp: FastMCP):
|
|
"""A caller-supplied value for a dependency parameter name is stripped.
|
|
|
|
A CallArgument that references the dependency parameter resolves the
|
|
dependency itself, never the caller's value.
|
|
"""
|
|
|
|
def get_role() -> str:
|
|
return "user"
|
|
|
|
def describe(role: str = CallArgument()) -> str:
|
|
return f"role={role}"
|
|
|
|
@mcp.prompt()
|
|
async def status(
|
|
topic: str,
|
|
role: str = Depends(get_role),
|
|
summary: str = Depends(describe),
|
|
) -> str:
|
|
return f"{topic}: {summary}"
|
|
|
|
result = await mcp.render_prompt("status", {"topic": "audit", "role": "admin"})
|
|
content = result.messages[0].content
|
|
assert isinstance(content, TextContent)
|
|
assert "role=user" in content.text
|
|
assert "admin" not in content.text
|