Python으로 커스텀 MCP 서버 만들기

사내에 굴러다니는 조회용 API와 운영 DB가 있다. 매번 "이 주문 상태 좀 봐줘" 같은 요청이 오면 누군가 쿼리를 던지고 결과를 복사해 전달한다. 이런 반복 작업을 Claude 같은 LLM이 직접 도구로 호출하게 만들면 어떨까. 그 접점이 MCP다. 최근 사내 도구를 MCP로 감싸는 작업을 몇 번 하면서 정리해 둔 내용을 공유한다.

결론부터 말하자면, Python mcp 패키지의 FastMCP를 쓰면 함수에 @mcp.tool() 하나만 붙여 도구를 만들 수 있다. 로컬 연결은 stdio 전송, 원격 배포는 Streamable HTTP 전송을 쓰고, HTTP에는 Bearer 토큰 미들웨어로 인증을 붙인다. DB 연결은 lifespan으로 한 번만 열어 도구에 주입한다. 아래에서 최소 서버부터 전송, 인증, 에러 처리까지 차례로 살펴본다.

1. MCP가 무엇이고 왜 필요한가

MCP(Model Context Protocol)는 LLM 애플리케이션과 외부 도구, 데이터 소스를 잇는 개방형 프로토콜이다. JSON-RPC 위에서 동작하며, 서버가 세 가지를 노출한다. 클라이언트가 읽는 파일형 데이터인 리소스(Resources), LLM이 호출하는 함수인 도구(Tools), 재사용 템플릿인 프롬프트(Prompts)다. 이 글은 가장 많이 쓰는 도구에 집중한다.

도구를 직접 만들면 좋은 점은, LLM이 어떤 함수를 언제 부를지 스스로 판단한다는 것이다. 우리는 함수 시그니처와 설명만 제공하면 된다. FastMCP는 Python 타입 힌트와 docstring을 읽어 도구 정의(JSON 스키마)를 자동 생성한다. 스키마를 손으로 쓸 일이 없다.

2. 전제 조건과 설치

mcp 패키지는 Python 3.10 이상에서 동작한다. 다만 최신 릴리스는 3.11 이상을 요구하니 설치 전에 버전을 확인한다.

$ python3 --version
Python 3.11.9

$ python3 -m venv .venv && source .venv/bin/activate
$ pip install "mcp[cli]"
...
Successfully installed mcp-1.x httpx-0.28.x starlette-0.x uvicorn-0.x ...

[cli] 추가 옵션은 mcp dev, mcp run 같은 명령 도구를 함께 설치한다. 테스트 때 유용하니 같이 받아두는 편이 낫다. 버전 번호는 설치 시점에 따라 다를 수 있다.

3. 최소 서버 만들기 (stdio)

server.py를 만들고 도구 두 개를 붙여본다.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("internal-tools")

@mcp.tool()
def add(a: int, b: int) -> int:
    """두 정수를 더한다."""
    return a + b

@mcp.tool()
async def whoami() -> str:
    """현재 서버가 실행 중인 호스트명을 돌려준다."""
    import socket
    return socket.gethostname()

if __name__ == "__main__":
    mcp.run(transport="stdio")

동기 함수든 async 함수든 똑같이 @mcp.tool()만 붙이면 된다. docstring 첫 줄이 도구 설명이 되고, 인자의 타입 힌트가 입력 스키마가 된다.

잘 떴는지는 MCP Inspector로 확인한다. mcp dev가 서버를 띄우고 브라우저 검사 도구를 함께 연다.

$ mcp dev server.py
Starting MCP inspector...
⚙️ Proxy server listening on 127.0.0.1:6277
🔍 MCP Inspector is up and running at http://127.0.0.1:6274

브라우저에서 Tools 탭을 열면 add와 whoami가 보인다. 여기서 직접 인자를 넣어 호출해보면 서버가 응답하는지 바로 확인할 수 있다.

4. 사내 DB와 API를 도구로 붙이기

실제로 쓸모 있으려면 DB나 내부 API를 물려야 한다. 매 도구 호출마다 커넥션을 새로 여는 건 낭비다. lifespan으로 서버가 뜰 때 한 번 연결하고, 각 도구에서 Context를 통해 꺼내 쓴다.

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass

import asyncpg
from mcp.server.fastmcp import FastMCP, Context

@dataclass
class AppCtx:
    pool: asyncpg.Pool

@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppCtx]:
    pool = await asyncpg.create_pool(dsn="postgresql://app@localhost/sarc")
    try:
        yield AppCtx(pool=pool)
    finally:
        await pool.close()

mcp = FastMCP("internal-tools", lifespan=lifespan)

@mcp.tool()
async def order_status(order_id: int, ctx: Context) -> str:
    """주문 번호로 현재 배송 상태를 조회한다.

    Args:
        order_id: 조회할 주문 번호
    """
    pool = ctx.request_context.lifespan_context.pool
    row = await pool.fetchrow(
        "SELECT status, updated_at FROM orders WHERE id = $1", order_id
    )
    if row is None:
        return f"주문 {order_id}을(를) 찾지 못했다."
    return f"상태: {row['status']}, 갱신: {row['updated_at']:%Y-%m-%d %H:%M}"

도구 함수의 마지막 인자로 ctx: Context를 선언하면 FastMCP가 주입해준다. 이 인자는 LLM에 노출되는 입력 스키마에는 들어가지 않는다. lifespan에서 만든 풀은 ctx.request_context.lifespan_context로 꺼낸다.

사내 REST API를 부르는 경우도 똑같다. lifespan에서 httpx.AsyncClient를 하나 만들어 재사용하면 커넥션을 아낄 수 있다.

처음엔 각 도구 안에서 asyncpg.connect()를 직접 불렀다가, 도구를 연달아 호출할 때 연결이 쌓여 금방 막혔다. 풀을 lifespan으로 올리고 나서 깔끔해졌다.

5. 에러는 어떻게 다뤄야 하나

도구 안에서 예외가 올라오면 FastMCP가 잡아서 isError가 붙은 도구 결과로 변환해 클라이언트에 돌려준다. 서버가 통째로 죽지는 않는다. 그래서 두 가지를 구분해야 한다.

LLM이 이해하고 대처할 수 있는 상황은 평범한 문자열로 돌려준다. 위 예시의 "주문을 찾지 못했다"가 그렇다. LLM은 이 응답을 읽고 사용자에게 다시 물어볼 수 있다.

잘못된 입력이나 처리 불가 상황은 예외를 던진다. 예외 메시지가 도구 에러로 전달된다.

@mcp.tool()
async def refund(order_id: int, amount: int, ctx: Context) -> str:
    """주문 금액 일부를 환불한다."""
    if amount <= 0:
        raise ValueError("환불 금액은 1 이상이어야 한다.")
    ...

주의할 점이 하나 있다. 내부 예외 메시지에 DB 접속 정보나 스택 트레이스가 그대로 실려 LLM에, 나아가 사용자에게 노출될 수 있다. 민감한 원본 예외는 try/except로 잡아 사람이 읽을 안전한 메시지로 바꿔 던지는 편이 안전하다.

6. stdio와 Streamable HTTP, 언제 뭘 쓰나

전송(transport)은 클라이언트와 서버가 메시지를 주고받는 통로다. 두 가지를 주로 쓴다.

stdio 전송

서버를 클라이언트가 자식 프로세스로 띄우고 표준 입출력으로 통신한다. Claude Desktop 같은 데스크톱 호스트에 로컬 도구를 붙일 때 이걸 쓴다. 네트워크 설정도, 인증도 필요 없다. 앞서 mcp.run(transport="stdio")로 이미 썼다.

Streamable HTTP 전송

서버를 독립 HTTP 서비스로 띄워 여러 사용자나 원격 클라이언트가 접속하게 한다. 사내 서버에 한 번 올려두고 팀이 공유하려면 이쪽이다. 실행만 바꾸면 된다.

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

$ python server.py
INFO:     Started server process [12345]
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

기본 접속 경로는 /mcp다. 호스트와 포트는 FastMCP("internal-tools", host="127.0.0.1", port=8000)로 지정한다.

정리하면 이렇다. 본인 노트북에서 Claude Desktop에만 물릴 거라면 stdio, 사내에 공유 서비스로 올릴 거라면 Streamable HTTP를 쓴다.

⚠️ HTTP로 열 때 주의. 기본은 127.0.0.1 바인딩이다. 0.0.0.0으로 외부에 열면 반드시 인증을 붙이고, DNS 리바인딩 공격을 막기 위해 Origin 헤더 검증도 함께 둬야 한다.

7. Bearer 토큰으로 인증 붙이기

HTTP로 열었으면 인증은 선택이 아니다. FastMCP는 streamable_http_app()으로 Starlette ASGI 앱을 돌려준다. 여기에 미들웨어를 얹어 토큰을 검사한다.

import os
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

API_TOKEN = os.environ["MCP_API_TOKEN"]

class BearerAuth(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        auth = request.headers.get("authorization", "")
        if auth != f"Bearer {API_TOKEN}":
            return JSONResponse({"error": "unauthorized"}, status_code=401)
        return await call_next(request)

app = mcp.streamable_http_app()
app.add_middleware(BearerAuth)

토큰은 코드에 박지 말고 환경 변수로 넘긴다. 실행은 uvicorn에 ASGI 앱을 지정한다.

$ MCP_API_TOKEN=s3cret uvicorn server:app --host 0.0.0.0 --port 8000
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

토큰 없이 접속하면 401이 떨어진다.

$ curl -s -o /dev/null -w "%{http_code}\n" -X POST http://localhost:8000/mcp
401

💡 OAuth 2.0 기반 인증이 필요하면 SDK에 토큰 검증기(TokenVerifier)와 인증 설정을 붙이는 방식이 따로 있다. 사내용으로 고정 토큰이면 위 미들웨어로 충분하다.

8. 동작 확인

Streamable HTTP 서버는 JSON-RPC로 말한다. 첫 요청은 initialize다. 응답이 SSE(text/event-stream)로 오므로 Accept 헤더에 두 타입을 모두 넣어야 한다.

$ curl -i -X POST http://localhost:8000/mcp \
  -H "Authorization: Bearer s3cret" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"curl","version":"1.0"}}}'

HTTP/1.1 200 OK
mcp-session-id: 3f2a...
content-type: text/event-stream

event: message
data: {"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-06-18","serverInfo":{"name":"internal-tools","version":"1.x"}}}

응답 헤더의 mcp-session-id를 이후 요청에 그대로 실어 보내면 같은 세션이 이어진다. Accept 헤더에 text/event-stream을 빼먹으면 406이 떨어지니 이 부분에서 한 번 걸리기 쉽다.

Claude Desktop 같은 로컬 호스트에 stdio 서버를 붙이려면 설정 파일에 등록한다.

{
  "mcpServers": {
    "internal-tools": {
      "command": "/path/to/.venv/bin/python",
      "args": ["/path/to/server.py"]
    }
  }
}

절대 경로로 넣어야 한다. 가상환경의 python을 가리키지 않으면 mcp 패키지를 못 찾아 서버가 뜨지 않는다.

9. 마무리

정리하면, 도구는 @mcp.tool()로 함수에 붙이고, DB 연결은 lifespan으로 한 번 열어 Context로 주입한다. 로컬이면 stdio, 사내 공유 서비스면 Streamable HTTP로 띄우고 HTTP에는 Bearer 토큰 미들웨어를 붙인다. 예외는 FastMCP가 도구 에러로 변환해주되, 민감 정보가 메시지에 실려 나가지 않도록 감싸준다. 사내에 흩어진 조회용 API와 쿼리를 이 패턴으로 하나씩 도구로 묶어두면, 반복 요청을 LLM이 직접 처리하게 만들 수 있다.