Spaces:
Running
Running
File size: 4,038 Bytes
c151bc3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | import httpx
import pytest
from fastapi import FastAPI, HTTPException, Response
from fastapi.responses import PlainTextResponse
from httpx import ASGITransport, AsyncClient
from pydantic import BaseModel
from fastmcp.server.openapi import (
FastMCPOpenAPI,
MCPType,
RouteMap,
)
class User(BaseModel):
id: int
name: str
active: bool
class UserCreate(BaseModel):
name: str
active: bool
@pytest.fixture
def users_db() -> dict[int, User]:
return {
1: User(id=1, name="Alice", active=True),
2: User(id=2, name="Bob", active=True),
3: User(id=3, name="Charlie", active=False),
}
# route maps for GET requests
# use these to create components of all types instead of just tools
GET_ROUTE_MAPS = [
# GET requests with path parameters go to ResourceTemplate
RouteMap(
methods=["GET"],
pattern=r".*\{.*\}.*",
mcp_type=MCPType.RESOURCE_TEMPLATE,
),
# GET requests without path parameters go to Resource
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
]
@pytest.fixture
def fastapi_app(users_db: dict[int, User]) -> FastAPI:
app = FastAPI(title="FastAPI App")
@app.get("/users", tags=["users", "list"])
async def get_users() -> list[User]:
"""Get all users."""
return sorted(users_db.values(), key=lambda x: x.id)
@app.get("/search", tags=["search"])
async def search_users(
name: str | None = None, active: bool | None = None, min_id: int | None = None
) -> list[User]:
"""Search users with optional filters."""
results = list(users_db.values())
if name is not None:
results = [u for u in results if name.lower() in u.name.lower()]
if active is not None:
results = [u for u in results if u.active == active]
if min_id is not None:
results = [u for u in results if u.id >= min_id]
return sorted(results, key=lambda x: x.id)
@app.get("/users/{user_id}", tags=["users", "detail"])
async def get_user(user_id: int) -> User | None:
"""Get a user by ID."""
return users_db.get(user_id)
@app.get("/users/{user_id}/{is_active}", tags=["users", "detail"])
async def get_user_active_state(user_id: int, is_active: bool) -> User | None:
"""Get a user by ID and filter by active state."""
user = users_db.get(user_id)
if user is not None and user.active == is_active:
return user
return None
@app.post("/users", tags=["users", "create"])
async def create_user(user: UserCreate) -> User:
"""Create a new user."""
user_id = max(users_db.keys()) + 1
new_user = User(id=user_id, **user.model_dump())
users_db[user_id] = new_user
return new_user
@app.patch("/users/{user_id}/name", tags=["users", "update"])
async def update_user_name(user_id: int, name: str) -> User:
"""Update a user's name."""
user = users_db.get(user_id)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
user.name = name
return user
@app.get("/ping", response_class=PlainTextResponse)
async def ping() -> str:
"""Ping the server."""
return "pong"
@app.get("/ping-bytes")
async def ping_bytes() -> Response:
"""Ping the server and get a bytes response."""
return Response(content=b"pong")
return app
@pytest.fixture
def api_client(fastapi_app: FastAPI) -> AsyncClient:
"""Create a pre-configured httpx client for testing."""
return AsyncClient(transport=ASGITransport(app=fastapi_app), base_url="http://test")
@pytest.fixture
async def fastmcp_openapi_server(
fastapi_app: FastAPI, api_client: httpx.AsyncClient
) -> FastMCPOpenAPI:
openapi_spec = fastapi_app.openapi()
return FastMCPOpenAPI(
openapi_spec=openapi_spec,
client=api_client,
name="Test App",
route_maps=GET_ROUTE_MAPS,
)
|