contracts.py

import asyncio
from fastmcp import FastMCP, Client
from fastmcp.exceptions import ToolError
from pydantic import BaseModel, Field

mcp = FastMCP("contracts", mask_error_details=True)
calls = []
class Issue(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    priority: int = Field(default=1, ge=0, le=3)
class IssueResult(BaseModel):
    number: int
    title: str

@mcp.tool
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a+b
@mcp.tool
def stats() -> dict:
    """Return demonstration counters."""
    return {"calls": 3}
@mcp.tool
def unannotated():
    """Return an object without declaring an output schema."""
    return {"calls": 3}
@mcp.tool
def create_issue(issue: Issue) -> IssueResult:
    """Create a local demonstration issue; fail deliberately for title offline."""
    calls.append(issue.title)
    if issue.title == "offline":
        raise ToolError("Issue service unavailable; no issue created in this fixture")
    return IssueResult(number=42, title=issue.title)

async def main():
    async with Client(mcp) as client:
        tools = {t.name:t for t in await client.list_tools()}
        for name,args in [("add",{"a":2,"b":3}),("stats",{}),("unannotated",{}),("create_issue",{"issue":{"title":"ticket-8821"}})]:
            result = await client.call_tool(name,args)
            print(name, "schema=",tools[name].output_schema,"structured=",result.structured_content,"data=",result.data)
        before = len(calls)
        invalid = await client.call_tool("create_issue",{"issue":{"title":""}},raise_on_error=False)
        assert invalid.is_error and len(calls)==before
        failed = await client.call_tool("create_issue",{"issue":{"title":"offline"}},raise_on_error=False)
        assert failed.is_error
        okay = await client.call_tool("add",{"a":2,"b":3})
        assert okay.data == 5
        print("invalid blocked before handler; controlled failure; next call succeeds")

if __name__ == "__main__":
    asyncio.run(main())