MCP 서버 개발 실전 가이드 — 사내 API와 DB를 도메인 전문 에이전트로 전환하기

[LLM Host / MCP Client]  ←→  [MCP Server]  ←→  [Domain Data / APIs]
  (Claude, Cursor 등)      (당신이 만드는 것)    (사내 DB, REST API 등)

# stdio 방식 (로컬 프로세스)
claude-desktop ──stdin/stdout──> python mcp_server.py
 
# SSE 방식 (원격 HTTP)
agent ──HTTP GET /sse──> https://mcp.yourcompany.com

pip install "mcp[cli]" httpx

# server.py
import asyncio
import os
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import httpx
 
app = Server("company-issue-tracker")
 
BASE_URL = "https://issues.yourcompany.com/api/v1"
 
 
def _auth_headers() -> dict:
    token = os.environ["ISSUE_TRACKER_TOKEN"]  # 환경 변수에서 토큰 로드
    return {"Authorization": f"Bearer {token}"}
 
 
@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_issue",
            description="이슈 트래커에서 특정 이슈 상세 정보를 조회합니다",
            inputSchema={
                "type": "object",
                "properties": {
                    "issue_id": {
                        "type": "string",
                        "description": "조회할 이슈 ID (예: PROJ-123)"
                    }
                },
                "required": ["issue_id"]
            }
        ),
        types.Tool(
            name="search_issues",
            description="키워드로 이슈를 검색합니다",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "status": {
                        "type": "string",
                        "enum": ["open", "closed", "all"]
                    }
                },
                "required": ["query"]
            }
        )
    ]
 
 
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    try:
        async with httpx.AsyncClient() as client:
            if name == "get_issue":
                resp = await client.get(
                    f"{BASE_URL}/issues/{arguments['issue_id']}",
                    headers=_auth_headers()
                )
                resp.raise_for_status()
                return [types.TextContent(type="text", text=resp.text)]
 
            elif name == "search_issues":
                params = {"q": arguments["query"]}
                if "status" in arguments:
                    params["status"] = arguments["status"]
                resp = await client.get(
                    f"{BASE_URL}/issues/search",
                    params=params,
                    headers=_auth_headers()
                )
                resp.raise_for_status()
                return [types.TextContent(type="text", text=resp.text)]
 
            else:
                return [types.TextContent(
                    type="text",
                    text=f"오류: '{name}'은 지원하지 않는 Tool입니다."
                )]
 
    except httpx.HTTPStatusError as e:
        return [types.TextContent(
            type="text",
            text=f"API 오류 ({e.response.status_code}): {e.response.text}"
        )]
    except Exception as e:
        return [types.TextContent(type="text", text=f"예상치 못한 오류: {str(e)}")]
 
 
async def main():
    async with stdio_server() as streams:
        await app.run(*streams, app.create_initialization_options())
 
 
if __name__ == "__main__":
    asyncio.run(main())

{
  "mcpServers": {
    "issue-tracker": {
      "command": "python",
      "args": ["/path/to/server.py"],
      "env": {
        "ISSUE_TRACKER_TOKEN": "your-token-here"
      }
    }
  }
}

pnpm add @modelcontextprotocol/sdk pg
pnpm add -D @types/pg tsx

{
  "type": "module",
  "scripts": {
    "start": "tsx server.ts"
  }
}

// server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { Pool } from "pg";
 
const server = new Server(
  { name: "db-schema-provider", version: "1.0.0" },
  { capabilities: { resources: {} } }
);
 
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
 
// 사용 가능한 Resource 목록 반환
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  const { rows } = await pool.query(`
    SELECT table_name
    FROM information_schema.tables
    WHERE table_schema = 'public'
    ORDER BY table_name
  `);
 
  return {
    resources: rows.map(row => ({
      uri: `db-schema://${row.table_name}`,
      name: `${row.table_name} 테이블 스키마`,
      mimeType: "application/json",
      description: `${row.table_name} 테이블의 컬럼 구조 및 타입`
    }))
  };
});
 
// 특정 Resource 내용 반환
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const tableName = request.params.uri.replace("db-schema://", "");
 
  const { rows } = await pool.query(`
    SELECT
      column_name,
      data_type,
      is_nullable,
      column_default
    FROM information_schema.columns
    WHERE table_schema = 'public' AND table_name = $1
    ORDER BY ordinal_position
  `, [tableName]);
 
  return {
    contents: [{
      uri: request.params.uri,
      mimeType: "application/json",
      text: JSON.stringify({ table: tableName, columns: rows }, null, 2)
    }]
  };
});
 
// 종료 시 DB 연결 풀 정리
process.on("SIGINT", async () => { await pool.end(); process.exit(0); });
process.on("SIGTERM", async () => { await pool.end(); process.exit(0); });
 
const transport = new StdioServerTransport();
await server.connect(transport);

# 나쁜 예 — 예외가 던져지면 에이전트 파이프라인 전체가 중단된다
raise ValueError("API 호출 실패")
 
# 좋은 예 — LLM이 상황을 인지하고 다음 행동을 스스로 결정한다
return [types.TextContent(
    type="text",
    text="오류: API 호출 실패 (503). 잠시 후 다시 시도하세요."
)]