Python 50줄로 MCP 서버를 만든 방법 (OpenAPI에서 자동 생성)

발행: (2025년 12월 6일 오전 10:50 GMT+9)
5 min read
원문: Dev.to

Source: Dev.to

Cover image for How I Built an MCP Server in 50 Lines of Python (Auto-Generated from OpenAPI)

What We’re Building

MCP 서버는 다음을 수행합니다:

  • ✅ 모든 OpenAPI 사양에서 도구를 자동 생성
  • ✅ 요청당 인증(API 키) 처리
  • ✅ Claude Desktop, Cursor 및 모든 MCP 클라이언트와 호환
  • ✅ 헬스 체크가 포함된 표준 웹 서비스로 배포

결과: AI 어시스턴트가 자연어를 통해 LinkedIn 작업(댓글 달기, 프로필 조회, 참여 관리 등)을 실행할 수 있습니다.

The Tech Stack

ComponentPurpose
FastMCPOpenAPI 지원이 포함된 MCP 서버 프레임워크
FastAPI미들웨어 및 헬스 체크를 위한 웹 프레임워크
httpx이벤트 훅을 지원하는 비동기 HTTP 클라이언트
ContextVar스레드‑안전한 요청당 상태 관리

The Architecture

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Claude Desktop │────▶│   MCP Server    │────▶│  LinkedIn API   │
│  / Cursor / AI  │     │  (FastMCP)      │     │  (OpenAPI)      │
└─────────────────┘     └─────────────────┘     └─────────────────┘
        │                       │
        │   ?apiKey=xxx         │  Authorization: Bearer xxx
        └───────────────────────┘
              API Key Flow

중간 레이어는 OpenAPI 사양을 읽어 모든 엔드포인트를 MCP 도구로 자동 노출합니다.

The Complete Code

1. Setting Up Per‑Request Authentication

from contextvars import ContextVar
import httpx

# Store API key per request using context variables (thread‑safe)
api_key_context: ContextVar[str] = ContextVar('api_key', default='')

# Event hook to inject Authorization header from context
async def add_auth_header(request: httpx.Request):
    """
    Async event hook that injects the Authorization header 
    from the context variable. Called for every outgoing request.
    """
    api_key = api_key_context.get()
    if api_key:
        request.headers["Authorization"] = f"Bearer {api_key}"

Why ContextVar? 전역 변수와 달리 ContextVar는 각 비동기 컨텍스트마다 별도 값을 유지하므로 동시 요청 간 키 누출을 방지합니다.

2. Creating the HTTP Client with Event Hooks

api_client = httpx.AsyncClient(
    base_url="https://api.connectsafely.ai/linkedin",
    timeout=300.0,  # 5‑minute timeout for long operations
    event_hooks={"request": [add_auth_header]}
)

event_hooks 패턴을 사용하면 모든 요청에 자동으로 올바른 Authorization 헤더가 추가됩니다.

3. Loading the OpenAPI Spec

import sys
import httpx

try:
    print("Loading OpenAPI spec...")
    openapi_spec = httpx.get(
        "https://api.connectsafely.ai/linkedin/openapi.json",
        timeout=5.0
    ).json()
    print("OpenAPI spec loaded successfully!")
except httpx.ConnectError:
    print("ERROR: Could not connect to API")
    sys.exit(1)
except Exception as e:
    print(f"ERROR: Failed to load OpenAPI spec: {e}")
    sys.exit(1)

Benefits:

  • Zero manual tool definitions – endpoints become tools automatically
  • Always in sync – updating the API updates MCP tools automatically
  • Self‑documenting – tool descriptions come from OpenAPI

4. The Magic: FastMCP from OpenAPI

from fastmcp import FastMCP

# Create the MCP server from the OpenAPI specification
mcp = FastMCP.from_openapi(
    openapi_spec=openapi_spec,
    client=api_client,
    name="LinkedIn API Server"
)

한 줄만으로 전체 OpenAPI 사양을 MCP 도구로 변환하며, 각 도구는 적절한 파라미터 검증, 타입 힌트 및 문서를 갖습니다.

5. FastAPI Integration for Production Features

from fastapi import FastAPI, Request

# Create MCP ASGI app
mcp_app = mcp.http_app(path='/')

# Create FastAPI app with MCP lifespan
app = FastAPI(lifespan=mcp_app.lifespan)

@app.get("/health")
async def health_check():
    """Health check endpoint for Docker and load balancers."""
    return {"status": "healthy"}

FastAPI는 헬스 체크, 커스텀 미들웨어 및 추가 REST 엔드포인트를 제공해 줍니다.

6. API Key Extraction Middleware

@app.middleware("http")
async def extract_api_key(request: Request, call_next):
    """
    Middleware to extract API key from query parameter 
    or Authorization header. Supports both formats:
    - ?apiKey=your-key
    - Authorization: Bearer your-key
    """
    # Extract from query parameter first
    api_key = request.query_params.get("apiKey", "")

    if not api_key:
        # Fallback to Authorization header
        auth_header = request.headers.get("Authorization", "")
        if auth_header.startswith("Bearer "):
            api_key = auth_header.split(" ")[1]

    # Store in context for downstream use
    api_key_context.set(api_key)

    response = await call_next(request)
    return response

# Mount MCP server at root
app.mount("/", mcp_app)

미들웨어는 각 요청에서 API 키를 추출해 ContextVar에 저장하고, 이후 httpx 훅이 이를 읽어 사용합니다.

7. Running the Server

if __name__ == "__main__":
    import uvicorn

    print("\nStarting MCP server on http://0.0.0.0:3011")
    print("Connect with: http://localhost:3011?apiKey=your-api-key")

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

8. Connecting to Claude Desktop

claude_desktop_config.json에 다음을 추가합니다:

{
  "mcpServers": {
    "connectsafely": {
      "url": "http://localhost:3011?apiKey=YOUR_API_KEY"
    }
  }
}

이제 Claude는 자연어를 통해 LinkedIn 자동화 작업을 실행할 수 있습니다. 예시:

“Post a comment on the latest post from @naval saying ‘Great insight on leverage!‘”

Claude는 다음 순서대로 수행합니다:

  1. 사용자의 게시물을 검색
  2. 최신 게시물을 찾음
  3. MCP‑생성 도구를 통해 댓글을 게시

Why This Pattern Is Powerful

1. Zero Duplication

Your REST API and MCP tools share the same OpenAPI source of truth. Updating the spec updates both automatically.

Back to Blog

관련 글

더 보기 »

3계층 Terraform 아키텍처

전제 조건: Terraform ≥ 1.0, S3 버킷, DynamoDB 테이블, KMS 키, VPC, 서브넷, IGW, route tables를 생성할 수 있는 권한이 있는 AWS 자격 증명.