"""Example FastAPI application for testing the MCP server."""
from typing import List, Optional
from fastapi import FastAPI, Depends, HTTPException, Query, Path
from pydantic import BaseModel, Field
# Models
class UserBase(BaseModel):
"""Base user model."""
email: str = Field(..., description="User's email address")
username: str = Field(..., min_length=3, max_length=50)
full_name: Optional[str] = None
class UserCreate(UserBase):
"""Model for creating a new user."""
password: str = Field(..., min_length=8)
class User(UserBase):
"""User model with ID."""
id: int
is_active: bool = True
model_config = {"json_schema_extra": {
"example": {
"id": 1,
"email": "john@example.com",
"username": "johndoe",
"full_name": "John Doe",
"is_active": True,
}
}}
class Item(BaseModel):
"""Item model."""
id: int
title: str
description: Optional[str] = None
owner_id: int
# Dependencies
def get_current_user() -> User:
"""Dependency to get the current authenticated user."""
return User(
id=1,
email="user@example.com",
username="testuser",
full_name="Test User",
)
def verify_admin(current_user: User = Depends(get_current_user)) -> User:
"""Dependency to verify user is admin."""
# Simplified admin check
if current_user.id != 1:
raise HTTPException(status_code=403, detail="Not enough permissions")
return current_user
# App
app = FastAPI(
title="Example FastAPI Application",
description="A sample API for demonstrating the FastAPI MCP server",
version="1.0.0",
)
# Routes
@app.get("/", tags=["root"])
async def root():
"""Root endpoint."""
return {"message": "Welcome to the Example API"}
@app.get(
"/users",
response_model=List[User],
tags=["users"],
summary="List all users",
description="Retrieve a list of all users in the system",
)
async def list_users(
skip: int = Query(0, ge=0, description="Number of users to skip"),
limit: int = Query(10, ge=1, le=100, description="Maximum number of users to return"),
):
"""List all users with pagination."""
return [
User(id=1, email="user1@example.com", username="user1", full_name="User One"),
User(id=2, email="user2@example.com", username="user2", full_name="User Two"),
]
@app.get(
"/users/{user_id}",
response_model=User,
tags=["users"],
summary="Get a user by ID",
)
async def get_user(
user_id: int = Path(..., ge=1, description="The ID of the user to retrieve"),
):
"""Get a specific user by their ID."""
return User(
id=user_id,
email=f"user{user_id}@example.com",
username=f"user{user_id}",
full_name=f"User {user_id}",
)
@app.post(
"/users",
response_model=User,
status_code=201,
tags=["users"],
summary="Create a new user",
)
async def create_user(user: UserCreate):
"""Create a new user account."""
return User(
id=999,
email=user.email,
username=user.username,
full_name=user.full_name,
)
@app.put(
"/users/{user_id}",
response_model=User,
tags=["users"],
summary="Update a user",
dependencies=[Depends(get_current_user)],
)
async def update_user(
user_id: int = Path(..., ge=1),
user: UserBase = ...,
):
"""Update an existing user's information."""
return User(
id=user_id,
email=user.email,
username=user.username,
full_name=user.full_name,
)
@app.delete(
"/users/{user_id}",
status_code=204,
tags=["users"],
summary="Delete a user",
dependencies=[Depends(verify_admin)],
)
async def delete_user(user_id: int = Path(..., ge=1)):
"""Delete a user (admin only)."""
return None
@app.get(
"/users/{user_id}/items",
response_model=List[Item],
tags=["users", "items"],
summary="Get user's items",
)
async def get_user_items(
user_id: int = Path(..., ge=1),
current_user: User = Depends(get_current_user),
):
"""Get all items belonging to a specific user."""
return [
Item(id=1, title="Item 1", description="First item", owner_id=user_id),
Item(id=2, title="Item 2", description="Second item", owner_id=user_id),
]
@app.get(
"/items",
response_model=List[Item],
tags=["items"],
summary="List all items",
)
async def list_items(
search: Optional[str] = Query(None, description="Search query for items"),
owner_id: Optional[int] = Query(None, description="Filter by owner ID"),
):
"""List all items with optional filtering."""
return [
Item(id=1, title="Item 1", description="First item", owner_id=1),
Item(id=2, title="Item 2", description="Second item", owner_id=2),
]
@app.post(
"/items",
response_model=Item,
status_code=201,
tags=["items"],
summary="Create a new item",
dependencies=[Depends(get_current_user)],
)
async def create_item(
item: Item,
current_user: User = Depends(get_current_user),
):
"""Create a new item."""
return item
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
