[MCP] FastAPI-MCP

이글을 보시는 분들은 MCP를 개발해보셨을거같아 MCP(model context protocal) 설명을 하지 않을것이다. 

 

FastAPI+MCP 를 사용한 openAPI 구조를 쉽게 구현할수가 있다.

즉 FASTAPI 로 작성한 API 서버의 앤드포인트를 MCP 도구로 자동 변환해주는 라이브러리다.

 요청/응답 모델의 Swagger 문서까지도 그도로 유지해준다. 

기존에 FastAPI 로 구성된 부분이 대부분이라면 쉽게 변경해주는 모듈이 매력적이다. 

글내용과 다르지만 openwebui 를 사용하시는 분들이라면
OpenWebUI는 HTTP REST API + OpenAPI(JSON 스키마) 방식만 지원 하여 
도구연결을 MCPO 방식으로 연결을 많이 해서 사용하실거다 

하지만 *FastAPI + MCP를 조합(Bridge) 방식으로 HTTP API 화 해서 사용하는 방식도 있다. 

여러 방법으로 개발 하고 있는 중이라 다음글에 관련해서 공유하겠습니다.

 

오늘은 FASTAPI_MCP로 바로 연결해서 사용할수있는 방법은 아래 참고하세요.

 

 

https://github.com/tadata-org/fastapi_mcp/tree/main

GitHub - tadata-org/fastapi_mcp: Expose your FastAPI endpoints as Model Context Protocol (MCP) tools, with Auth!

 

MCP 형식을 

• Authentication built in, using your existing FastAPI dependencies!
• FastAPI-native: Not just another OpenAPI -> MCP converter
• Zero/Minimal configuration required - just point it at your FastAPI app and it works
• Preserving schemas of your request models and response models
• Preserve documentation of all your endpoints, just as it is in Swagger
• Flexible deployment - Mount your MCP server to the same app, or deploy separately
• ASGI transport - Uses FastAPI's ASGI interface directly for efficient communication

우선 설치를 해주고

 아래는 일부 경험했던 부분과 Git에서 친절하게 예시가 있어서 가져왔다. 

패키지설치
pip install fastapi-mcp

git 기본 예시
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP

app = FastAPI()

mcp = FastApiMCP(app)

# Mount the MCP server directly to your FastAPI app
mcp.mount()



사용자 예시
from fastapi_mcp import tool

@tool(
    name="get_weather",
    description="도시의 현재 날씨 정보를 반환합니다.",
)
def get_weather(city: str) -> dict:
    # (예시) 외부 API 호출 등
    return {"city": city, "weather": "맑음"}
    
    
    """
This example shows how to describe the full response schema instead of just a response example.
"""
from examples.shared.apps.items import app # The FastAPI app
from examples.shared.setup import setup_logging

from fastapi_mcp import FastApiMCP

setup_logging()

# Add MCP server to the FastAPI app
mcp = FastApiMCP(
    app,
    name="Item API MCP",
    description="MCP server for the Item API",
    describe_full_response_schema=True,  # Describe the full response JSON-schema instead of just a response example
    describe_all_responses=True,  # Describe all the possible responses instead of just the success (2XX) response
)

mcp.mount()

if __name__ == "__main__":
    import uvicorn
    
    uvicorn.run(app, host="0.0.0.0", port=8000)
    
    
    
    
 """
작업 ID와 태그를 필터링하여 노출 끝점을 사용자 지정하는 방법을 보여줍니다.
필터링에 관한 참고 사항:
- '포함_연산'과 '제외_연산'을 동시에 사용할 수 없습니다
- 'include_tags'와 'exclused_tags'를 동시에 사용할 수 없습니다
- 작업 필터링과 태그 필터링을 결합할 수 있습니다(예: 'include_operations'와 'include_tags'를 함께 사용)
- 필터를 결합할 때 탐욕스러운 접근 방식을 취합니다. 두 기준 중 하나에 일치하는 엔드포인트가 포함됩니다
"""
from examples.shared.apps.items import app # The FastAPI app
from examples.shared.setup import setup_logging

from fastapi_mcp import FastApiMCP

setup_logging()

# Examples demonstrating how to filter MCP tools by operation IDs and tags

# Filter by including specific operation IDs
include_operations_mcp = FastApiMCP(
    app,
    name="Item API MCP - Included Operations",
    include_operations=["get_item", "list_items"],
)

# Filter by excluding specific operation IDs
exclude_operations_mcp = FastApiMCP(
    app,    
    name="Item API MCP - Excluded Operations",
    exclude_operations=["create_item", "update_item", "delete_item"],
)

# Filter by including specific tags
include_tags_mcp = FastApiMCP(
    app,
    name="Item API MCP - Included Tags",
    include_tags=["items"],
)

# Filter by excluding specific tags
exclude_tags_mcp = FastApiMCP(
    app,
    name="Item API MCP - Excluded Tags",
    exclude_tags=["search"],
)

# Combine operation IDs and tags (include mode)
combined_include_mcp = FastApiMCP(
    app,
    name="Item API MCP - Combined Include",
    include_operations=["delete_item"],
    include_tags=["search"],
)

# Mount all MCP servers with different paths
include_operations_mcp.mount(mount_path="/include-operations-mcp")
exclude_operations_mcp.mount(mount_path="/exclude-operations-mcp")
include_tags_mcp.mount(mount_path="/include-tags-mcp")
exclude_tags_mcp.mount(mount_path="/exclude-tags-mcp")
combined_include_mcp.mount(mount_path="/combined-include-mcp")

if __name__ == "__main__":
    import uvicorn

    print("Server is running with multiple MCP endpoints:")
    print(" - /include-operations-mcp: Only get_item and list_items operations")
    print(" - /exclude-operations-mcp: All operations except create_item, update_item, and delete_item")
    print(" - /include-tags-mcp: Only operations with the 'items' tag")
    print(" - /exclude-tags-mcp: All operations except those with the 'search' tag")
    print(" - /combined-include-mcp: Operations with 'search' tag or delete_item operation")
    uvicorn.run(app, host="0.0.0.0", port=8000)   
    
    
    
    """
이 예제는 MCP 서버가 생성된 후 엔드포인트를 추가하는 경우 도구를 재등록하는 방법을 보여줍니다
"""
from examples.shared.apps.items import app # The FastAPI app
from examples.shared.setup import setup_logging

from fastapi_mcp import FastApiMCP

setup_logging()

mcp = FastApiMCP(app) # Add MCP server to the FastAPI app
mcp.mount() # MCP server


# This endpoint will not be registered as a tool, since it was added after the MCP instance was created
@app.get("/new/endpoint/", operation_id="new_endpoint", response_model=dict[str, str])
async def new_endpoint():
    return {"message": "Hello, world!"}


# But if you re-run the setup, the new endpoints will now be exposed.
mcp.setup_server()


if __name__ == "__main__":
    import uvicorn
    
    uvicorn.run(app, host="0.0.0.0", port=8000)
    
    
    
   """
This example shows how to mount the MCP server to a specific APIRouter, giving a custom mount path.
"""
from examples.shared.apps.items import app # The FastAPI app
from examples.shared.setup import setup_logging

from fastapi import APIRouter
from fastapi_mcp import FastApiMCP

setup_logging()

other_router = APIRouter(prefix="/other/route")    
app.include_router(other_router)

mcp = FastApiMCP(app)

# Mount the MCP server to a specific router.
# It will now only be available at `/other/route/mcp`
mcp.mount(other_router)


if __name__ == "__main__":
    import uvicorn
    
    uvicorn.run(app, host="0.0.0.0", port=8000) 
    
    
   """
예는 MCP 서버의 HTTP 클라이언트 타임아웃을 설정하는 방법을 보여줍니다.
응답하는 데 5초 이상 걸리는 API 엔드포인트가 있는 경우 타임아웃을 늘릴 수 있습니다.
"""
from examples.shared.apps.items import app # The FastAPI app
from examples.shared.setup import setup_logging

import httpx

from fastapi_mcp import FastApiMCP

setup_logging()


mcp = FastApiMCP(
    app,
    http_client=httpx.AsyncClient(timeout=20)
)
mcp.mount()


if __name__ == "__main__":
    import uvicorn
    
    uvicorn.run(app, host="0.0.0.0", port=8000) 
    
"""
이 예는 권한 부여 헤더에 유효한 토큰이 전달되지 않은 상태에서 요청을 거부하는 방법을 보여줍니다.

인증 헤더를 구성하려면 MCP 서버의 구성 파일이 다음과 같아야 합니다
```json
{
  "mcpServers": {
    "remote-example": {
      "command": "npx",
      "args": [
        "mcp-remote",
        "http://localhost:8000/mcp",
        "--header",
        "Authorization:${AUTH_HEADER}"
      ]
    },
    "env": {
      "AUTH_HEADER": "Bearer <your-token>"
    }
  }
}
```
"""
from examples.shared.apps.items import app # The FastAPI app
from examples.shared.setup import setup_logging

from fastapi import Depends
from fastapi.security import HTTPBearer

from fastapi_mcp import FastApiMCP, AuthConfig

setup_logging()

# Scheme for the Authorization header
token_auth_scheme = HTTPBearer()

# Create a private endpoint
@app.get("/private")
async def private(token = Depends(token_auth_scheme)):
    return token.credentials

# Create the MCP server with the token auth scheme
mcp = FastApiMCP(
    app,
    name="Protected MCP",
    auth_config=AuthConfig(
        dependencies=[Depends(token_auth_scheme)],
    ),
)

# Mount the MCP server
mcp.mount()


if __name__ == "__main__":
    import uvicorn
    
    uvicorn.run(app, host="0.0.0.0", port=8000)    
    
    
    
 from fastapi import FastAPI, Depends, HTTPException, Request, status
from pydantic_settings import BaseSettings
from typing import Any
import logging

from fastapi_mcp import FastApiMCP, AuthConfig

from examples.shared.auth import fetch_jwks_public_key
from examples.shared.setup import setup_logging


setup_logging()
logger = logging.getLogger(__name__)


class Settings(BaseSettings):
    """
    For this to work, you need an .env file in the root of the project with the following variables:
    AUTH0_DOMAIN=your-tenant.auth0.com
    AUTH0_AUDIENCE=https://your-tenant.auth0.com/api/v2/
    AUTH0_CLIENT_ID=your-client-id
    AUTH0_CLIENT_SECRET=your-client-secret
    """

    auth0_domain: str  # Auth0 domain, e.g. "your-tenant.auth0.com"
    auth0_audience: str  # Audience, e.g. "https://your-tenant.auth0.com/api/v2/"
    auth0_client_id: str
    auth0_client_secret: str

    @property
    def auth0_jwks_url(self):
        return f"https://{self.auth0_domain}/.well-known/jwks.json"

    @property
    def auth0_oauth_metadata_url(self):
        return f"https://{self.auth0_domain}/.well-known/openid-configuration"

    class Config:
        env_file = ".env"


settings = Settings()  # type: ignore


async def lifespan(app: FastAPI):
    app.state.jwks_public_key = await fetch_jwks_public_key(settings.auth0_jwks_url)
    logger.info(f"Auth0 client ID in settings: {settings.auth0_client_id}")
    logger.info(f"Auth0 domain in settings: {settings.auth0_domain}")
    logger.info(f"Auth0 audience in settings: {settings.auth0_audience}")
    yield


async def verify_auth(request: Request) -> dict[str, Any]:
    try:
        import jwt

        auth_header = request.headers.get("authorization", "")
        if not auth_header.startswith("Bearer "):
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authorization header")

        token = auth_header.split(" ")[1]

        header = jwt.get_unverified_header(token)

        # Check if this is a JWE token (encrypted token)
        if header.get("alg") == "dir" and header.get("enc") == "A256GCM":
            raise ValueError(
                "Token is encrypted, offline validation not possible. "
                "This is usually due to not specifying the audience when requesting the token."
            )

        # Otherwise, it's a JWT, we can validate it offline
        if header.get("alg") in ["RS256", "HS256"]:
            claims = jwt.decode(
                token,
                app.state.jwks_public_key,
                algorithms=["RS256", "HS256"],
                audience=settings.auth0_audience,
                issuer=f"https://{settings.auth0_domain}/",
                options={"verify_signature": True},
            )
            return claims

    except Exception as e:
        logger.error(f"Auth error: {str(e)}")

    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")


async def get_current_user_id(claims: dict = Depends(verify_auth)) -> str:
    user_id = claims.get("sub")

    if not user_id:
        logger.error("No user ID found in token")
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")

    return user_id


app = FastAPI(lifespan=lifespan)


@app.get("/api/public", operation_id="public")
async def public():
    return {"message": "This is a public route"}


@app.get("/api/protected", operation_id="protected")
async def protected(user_id: str = Depends(get_current_user_id)):
    return {"message": f"Hello, {user_id}!", "user_id": user_id}


# Set up FastAPI-MCP with Auth0 auth
mcp = FastApiMCP(
    app,
    name="MCP With Auth0",
    description="Example of FastAPI-MCP with Auth0 authentication",
    auth_config=AuthConfig(
        issuer=f"https://{settings.auth0_domain}/",
        authorize_url=f"https://{settings.auth0_domain}/authorize",
        oauth_metadata_url=settings.auth0_oauth_metadata_url,
        audience=settings.auth0_audience,
        client_id=settings.auth0_client_id,
        client_secret=settings.auth0_client_secret,
        dependencies=[Depends(verify_auth)],
        setup_proxies=True,
    ),
)

# Mount the MCP server
mcp.mount()


if __name__ == "__main__":
    import uvicorn

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