10_console_example.py

"""
Example demonstrating the MCP Web Console feature.

This example shows how to:
1. Create a FastAPI app with some endpoints
2. Add MCP server capabilities
3. Mount the Web Console for visualization and debugging

Access the console at: http://localhost:8000/mcp-console
"""

import sys
import os
import site
from pathlib import Path

user_site = site.getusersitepackages()
if user_site and user_site not in sys.path:
    sys.path.insert(0, user_site)

project_root = Path(__file__).parent.parent
if str(project_root) not in sys.path:
    sys.path.insert(0, str(project_root))

print("=" * 60)
print("Python Path Debug Info")
print("=" * 60)
print(f"Project root: {project_root}")
print(f"User site-packages: {user_site}")
print(f"Current file: {__file__}")
print()
print("sys.path:")
for i, p in enumerate(sys.path):
    print(f"  [{i}] {p}")
print()
print("Checking fastapi_mcp module location...")

try:
    import fastapi_mcp
    print(f"fastapi_mcp imported from: {fastapi_mcp.__file__}")
    print(f"fastapi_mcp version: {fastapi_mcp.__version__}")
    
    expected_server_py = project_root / 'fastapi_mcp' / 'server.py'
    actual_server_py = Path(fastapi_mcp.__file__).parent / 'server.py'
    
    print()
    print(f"Expected server.py: {expected_server_py}")
    print(f"Actual module path: {Path(fastapi_mcp.__file__).parent}")
    
    if str(expected_server_py.parent) != str(Path(fastapi_mcp.__file__).parent):
        print()
        print("WARNING: Importing from different location!")
        print("This may be an older installed version.")
    
    print()
    print("Checking FastApiMCP methods...")
    from fastapi_mcp import FastApiMCP
    methods = [m for m in dir(FastApiMCP) if not m.startswith('_')]
    print(f"FastApiMCP public methods: {methods}")
    
    if 'mount_console' not in methods:
        print()
        print("=" * 60)
        print("ERROR: 'mount_console' method not found!")
        print("=" * 60)
        print()
        print("This means Python is importing an older version of fastapi-mcp")
        print("that doesn't have the new console feature.")
        print()
        print(f"Expected location: {expected_server_py}")
        print(f"Actual location: {actual_server_py}")
        print()
        print("Solutions:")
        print("  1. Force reinstall local version:")
        print("     pip install -e . --user --force-reinstall --no-deps")
        print()
        print("  2. Or check what's in actual location:")
        print(f"     dir {actual_server_py.parent}")
        print()
        print("  3. Or remove the older version from site-packages:")
        print(f"     pip uninstall fastapi-mcp")
        print(f"     Then run: pip install -e . --user")
        sys.exit(1)
    else:
        print("✓ 'mount_console' method found!")
        
except ImportError as e:
    print()
    print("=" * 60)
    print(f"ERROR: Failed to import: {e}")
    print("=" * 60)
    print()
    print("This usually means dependencies are not installed or not in path.")
    print()
    print("Try installing dependencies:")
    print("  pip install fastapi uvicorn pydantic mcp httpx --user")
    sys.exit(1)

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from typing import List, Optional


app = FastAPI(
    title="MCP Console Demo",
    description="A demo FastAPI app with MCP Web Console",
    version="1.0.0",
)


class User(BaseModel):
    id: int
    name: str
    email: str
    age: Optional[int] = None
    active: bool = True


class CreateUserRequest(BaseModel):
    name: str
    email: str
    age: Optional[int] = None


class UpdateUserRequest(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None
    age: Optional[int] = None
    active: Optional[bool] = None


users_db: dict[int, User] = {}


sample_users = [
    User(id=1, name="Alice Johnson", email="alice@example.com", age=30),
    User(id=2, name="Bob Smith", email="bob@example.com", age=25),
    User(id=3, name="Charlie Brown", email="charlie@example.com", age=35),
]
for user in sample_users:
    users_db[user.id] = user


@app.get("/users/", response_model=List[User], tags=["users"], operation_id="list_users")
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"),
    active_only: bool = Query(False, description="Only return active users"),
):
    """
    List all users with pagination support.

    Returns a paginated list of users from the database.
    """
    results = list(users_db.values())

    if active_only:
        results = [u for u in results if u.active]

    return results[skip : skip + limit]


@app.get("/users/{user_id}", response_model=User, tags=["users"], operation_id="get_user")
async def get_user(user_id: int):
    """
    Get a specific user by ID.

    Returns the user details if found, otherwise raises a 404 error.
    """
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[user_id]


@app.post("/users/", response_model=User, tags=["users"], operation_id="create_user")
async def create_user(user: CreateUserRequest):
    """
    Create a new user.

    Generates a new unique ID for the user and adds them to the database.
    Returns the created user with the assigned ID.
    """
    new_id = max(users_db.keys(), default=0) + 1
    new_user = User(
        id=new_id,
        name=user.name,
        email=user.email,
        age=user.age,
        active=True,
    )
    users_db[new_id] = new_user
    return new_user


@app.put("/users/{user_id}", response_model=User, tags=["users"], operation_id="update_user")
async def update_user(user_id: int, user: UpdateUserRequest):
    """
    Update an existing user.

    Only updates the fields that are provided in the request.
    Raises a 404 error if the user does not exist.
    """
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")

    existing = users_db[user_id]

    if user.name is not None:
        existing.name = user.name
    if user.email is not None:
        existing.email = user.email
    if user.age is not None:
        existing.age = user.age
    if user.active is not None:
        existing.active = user.active

    return existing


@app.delete("/users/{user_id}", tags=["users"], operation_id="delete_user")
async def delete_user(user_id: int):
    """
    Delete a user.

    Removes the user from the database.
    Raises a 404 error if the user does not exist.
    """
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")

    del users_db[user_id]
    return {"message": "User deleted successfully", "user_id": user_id}


@app.get("/users/search/", response_model=List[User], tags=["search"], operation_id="search_users")
async def search_users(
    q: Optional[str] = Query(None, description="Search query for name or email"),
    min_age: Optional[int] = Query(None, description="Minimum age filter"),
    max_age: Optional[int] = Query(None, description="Maximum age filter"),
    active: Optional[bool] = Query(None, description="Filter by active status"),
):
    """
    Search users with various filter options.

    Allows searching by name/email, and filtering by age range and active status.
    """
    results = list(users_db.values())

    if q:
        q = q.lower()
        results = [
            user for user in results
            if q in user.name.lower() or q in user.email.lower()
        ]

    if min_age is not None:
        results = [user for user in results if user.age is not None and user.age >= min_age]

    if max_age is not None:
        results = [user for user in results if user.age is not None and user.age <= max_age]

    if active is not None:
        results = [user for user in results if user.active == active]

    return results


@app.get("/health", tags=["system"], operation_id="health_check")
async def health_check():
    """
    Health check endpoint.

    Returns the current health status of the application.
    """
    return {
        "status": "healthy",
        "users_count": len(users_db),
    }


mcp = FastApiMCP(app)

mcp.mount_http()

mcp.mount_console(
    mount_path="/mcp-console",
    log_max_size=1000,
)

print("=" * 60)
print("FastAPI-MCP Console Demo")
print("=" * 60)
print()
print("Available endpoints:")
print("  - Swagger UI:  http://localhost:8000/docs")
print("  - MCP HTTP:    http://localhost:8000/mcp")
print("  - MCP Console: http://localhost:8000/mcp-console")
print()
print("Access the Web Console to:")
print("  - View all registered MCP tools")
print("  - See tool descriptions, parameters, and responses")
print("  - Test tool executions with custom parameters")
print("  - View real-time call logs with timings")
print()
print("=" * 60)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)