
Claude Agent에는 Read, Write, Bash, Glob 같은 내장 도구가 있습니다. 하지만 실제 프로젝트에서는 이것만으로 부족합니다:
MCP(Model Context Protocol)로 이런 커스텀 도구를 만들어서 Claude에 연결할 수 있습니다. Claude가 "이 사용자의 주문 내역을 조회해줘"라는 요청을 받으면, 내가 만든 get_orders 도구를 호출하는 식입니다.
이 글에서는 MCP 서버를 만드는 두 가지 방식(인프로세스 / 외부 프로세스)을 다루고, SDK가 내부적으로 어떻게 도구 호출을 라우팅하는지까지 분석합니다.
MCP(Model Context Protocol)는 AI 모델이 외부 도구를 호출하기 위한 표준 프로토콜입니다. JSON-RPC 기반이며, 핵심 흐름은 단순합니다:
Claude: "tools/list를 호출해서 사용 가능한 도구 목록을 알려줘"
서버: [{"name": "get_orders", "description": "주문 조회", ...}]
Claude: "tools/call로 get_orders를 호출해줘. input: {user_id: 123}"
서버: {"content": [{"type": "text", "text": "주문 3건: ..."}]}
Claude Agent SDK에서는 4가지 방식으로 MCP 서버를 연결할 수 있습니다:
| 방식 | 타입 | 서버 위치 | 특징 |
|---|---|---|---|
| SDK 인프로세스 | "sdk" | 같은 Python 프로세스 | IPC 없음, 앱 상태 직접 접근 |
| stdio | "stdio" | 로컬 별도 프로세스 | 가장 보편적 |
| SSE | "sse" | 원격 서버 | Server-Sent Events |
| HTTP | "http" | 원격 서버 | Streamable HTTP |
이 글에서는 SDK 인프로세스 방식을 중심으로 다루고, 외부 서버 연결도 설명합니다.
외부 MCP 서버는 별도 프로세스를 띄우고 stdin/stdout 또는 HTTP로 통신합니다. 인프로세스 서버는 같은 Python 프로세스 안에서 실행됩니다.
외부 서버: Claude CLI → (stdin/stdout) → 별도 프로세스
인프로세스: Claude CLI → (control protocol) → SDK → 같은 프로세스의 함수 호출
장점:
from claude_agent_sdk import tool
@tool("greet", "사용자에게 인사하기", {"name": str})
async def greet(args):
return {
"content": [
{"type": "text", "text": f"안녕하세요, {args['name']}님!"}
]
}
from claude_agent_sdk import create_sdk_mcp_server
server = create_sdk_mcp_server(
name="my-tools",
version="1.0.0",
tools=[greet],
)
from claude_agent_sdk import query, ClaudeAgentOptions
options = ClaudeAgentOptions(
permission_mode="bypassPermissions",
mcp_servers={"my-tools": server},
allowed_tools=["greet"], # MCP 도구도 이름으로 허용
)
async for message in query(prompt="홍길동에게 인사해줘", options=options):
...
Claude는 greet 도구를 인식하고, 적절한 시점에 호출합니다.
@tool(name, description, input_schema, annotations=None)
| 인자 | 타입 | 설명 |
|---|---|---|
name | str | 도구 이름 (Claude가 호출할 때 사용) |
description | str | 도구 설명 (Claude가 언제 사용할지 판단) |
input_schema | dict \| TypedDict \| JSON Schema | 입력 파라미터 정의 |
annotations | ToolAnnotations \| None | 도구 메타데이터 (선택) |
방식 1: 딕셔너리 (가장 간단)
@tool("add", "두 수를 더합니다", {"a": float, "b": float})
async def add(args):
result = args["a"] + args["b"]
return {"content": [{"type": "text", "text": f"결과: {result}"}]}
방식 2: TypedDict (복잡한 스키마)
from typing import TypedDict, Annotated
class SearchInput(TypedDict):
query: Annotated[str, "검색어"]
max_results: Annotated[int, "최대 결과 수"]
@tool("search", "데이터베이스 검색", SearchInput)
async def search(args):
# args["query"], args["max_results"] 사용
...
Annotated[type, "설명"]을 사용하면 각 파라미터에 설명을 붙일 수 있습니다. Claude가 도구를 올바르게 사용하는 데 도움을 줍니다.
방식 3: JSON Schema (완전한 제어)
@tool("create_user", "사용자 생성", {
"type": "object",
"properties": {
"name": {"type": "string", "description": "사용자 이름"},
"age": {"type": "integer", "minimum": 0, "maximum": 150},
"role": {"type": "string", "enum": ["admin", "user", "guest"]},
},
"required": ["name", "role"],
})
async def create_user(args):
...
JSON Schema를 직접 넘기면 enum, minimum, maximum 같은 제약 조건도 사용할 수 있습니다.
도구 함수는 반드시 content 키를 포함하는 딕셔너리를 반환해야 합니다:
# 텍스트 반환
return {
"content": [
{"type": "text", "text": "결과입니다."}
]
}
# 에러 반환
return {
"content": [
{"type": "text", "text": "사용자를 찾을 수 없습니다."}
],
"is_error": True
}
# 이미지 반환
return {
"content": [
{
"type": "image",
"data": base64_encoded_string,
"mimeType": "image/png",
}
]
}
# 여러 콘텐츠 혼합
return {
"content": [
{"type": "text", "text": "차트를 생성했습니다:"},
{"type": "image", "data": chart_base64, "mimeType": "image/png"},
]
}
지원하는 콘텐츠 타입:
| type | 설명 |
|---|---|
"text" | 텍스트 (text 필드) |
"image" | 이미지 (data + mimeType 필드) |
"resource_link" | 리소스 링크 (name, uri, description) |
"resource" | 내장 리소스 (resource.text) |
도구가 읽기전용인지, 파괴적인지 등을 명시할 수 있습니다.
from mcp.types import ToolAnnotations
@tool(
"delete_user",
"사용자 삭제",
{"user_id": int},
annotations=ToolAnnotations(
destructive=True, # 데이터를 삭제/수정함
readOnly=False, # 읽기만 하는 게 아님
openWorld=False, # 외부 네트워크 접근 안 함
),
)
async def delete_user(args):
...
| 어노테이션 | 기본값 | 의미 |
|---|---|---|
readOnly | False | True면 읽기만 수행 |
destructive | False | True면 데이터를 삭제하거나 수정 |
openWorld | False | True면 외부 네트워크에 접근 |
def create_sdk_mcp_server(
name: str,
version: str = "1.0.0",
tools: list[SdkMcpTool] | None = None,
) -> McpSdkServerConfig
내부적으로 이 함수는:
1. mcp.server.Server 인스턴스 생성
2. @server.list_tools()에 도구 목록 등록
3. @server.call_tool()에 도구 실행 핸들러 등록
4. McpSdkServerConfig(type="sdk", name=name, instance=server) 반환
반환된 config를 ClaudeAgentOptions.mcp_servers에 넣으면 됩니다.
SDK 인프로세스 서버는 외부 서버와 다른 경로로 동작합니다. 내부를 이해하면 디버깅이 쉬워집니다.
CLI가 직접 서버에 연결 → MCP 프로토콜로 직접 통신
(SDK는 관여하지 않음)
Claude가 도구 호출 결정
│
▼
CLI → SDK: control_request (mcp_message)
│ server_name, JSONRPC 메시지 전달
│
▼
SDK: _handle_sdk_mcp_request()
│
├── "tools/list" → server.list_tools() → 도구 목록 반환
└── "tools/call" → server.call_tool(name, args) → 핸들러 실행
│
▼
SDK → CLI: control_response (mcp_response)
│ JSONRPC 응답 전달
│
▼
CLI: 결과를 Claude에게 전달
핵심 차이는 CLI가 직접 서버에 연결하지 않고, SDK를 통해 제어 프로토콜로 라우팅한다는 것입니다. CLI에는 instance 필드가 제거된 메타데이터만 전달됩니다.
인프로세스가 아닌, 별도 프로세스나 원격 서버를 연결하는 방법입니다.
가장 보편적인 MCP 서버 연결 방식입니다. CLI가 subprocess를 띄워서 stdin/stdout으로 통신합니다.
options = ClaudeAgentOptions(
mcp_servers={
"postgres": {
"type": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres",
"postgresql://localhost:5432/mydb"],
},
"filesystem": {
"type": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem",
"/path/to/allowed/dir"],
},
}
)
| 필드 | 타입 | 설명 |
|---|---|---|
type | "stdio" | (생략 가능 — 기본값) |
command | str | 실행할 명령 |
args | list[str] | 명령 인자 (선택) |
env | dict[str, str] | 환경변수 (선택) |
원격 서버에 SSE로 연결합니다.
options = ClaudeAgentOptions(
mcp_servers={
"remote-tools": {
"type": "sse",
"url": "https://my-mcp-server.example.com/sse",
"headers": {
"Authorization": "Bearer my-token",
},
},
}
)
원격 서버에 HTTP로 연결합니다.
options = ClaudeAgentOptions(
mcp_servers={
"api-tools": {
"type": "http",
"url": "https://my-mcp-server.example.com/mcp",
"headers": {
"X-API-Key": "my-key",
},
},
}
)
mcp_servers에 딕셔너리 대신 JSON 파일 경로를 넣을 수도 있습니다.
# JSON 파일로 설정
options = ClaudeAgentOptions(
mcp_servers="/path/to/mcp-config.json"
)
mcp-config.json:
{
"mcpServers": {
"postgres": {
"type": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://..."]
}
}
}
Python 애플리케이션의 메모리 상태를 Claude가 조회할 수 있게 합니다.
# 앱 상태
users_db = {
1: {"name": "홍길동", "role": "admin", "orders": 5},
2: {"name": "김철수", "role": "user", "orders": 12},
}
@tool("get_user", "사용자 정보 조회", {"user_id": int})
async def get_user(args):
user = users_db.get(args["user_id"])
if not user:
return {
"content": [{"type": "text", "text": f"사용자 {args['user_id']}를 찾을 수 없습니다."}],
"is_error": True,
}
return {
"content": [{"type": "text", "text": json.dumps(user, ensure_ascii=False)}]
}
@tool("list_users", "전체 사용자 목록", {})
async def list_users(args):
summary = [f"ID {uid}: {u['name']} ({u['role']})" for uid, u in users_db.items()]
return {
"content": [{"type": "text", "text": "\n".join(summary)}]
}
server = create_sdk_mcp_server("user-db", tools=[get_user, list_users])
외부 API 호출을 Claude의 도구로 감쌉니다.
import httpx
@tool("get_weather", "날씨 조회", {
"city": Annotated[str, "도시 이름 (예: Seoul)"],
})
async def get_weather(args):
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.weatherapi.com/v1/current.json",
params={"key": API_KEY, "q": args["city"]},
)
if resp.status_code != 200:
return {
"content": [{"type": "text", "text": f"API 에러: {resp.status_code}"}],
"is_error": True,
}
data = resp.json()
temp = data["current"]["temp_c"]
condition = data["current"]["condition"]["text"]
return {
"content": [{"type": "text", "text": f"{args['city']}: {temp}°C, {condition}"}]
}
import math
@tool("add", "두 수를 더합니다", {"a": float, "b": float})
async def add(args):
result = args["a"] + args["b"]
return {"content": [{"type": "text", "text": str(result)}]}
@tool("multiply", "두 수를 곱합니다", {"a": float, "b": float})
async def multiply(args):
result = args["a"] * args["b"]
return {"content": [{"type": "text", "text": str(result)}]}
@tool("sqrt", "제곱근을 구합니다", {"n": float})
async def sqrt(args):
if args["n"] < 0:
return {"content": [{"type": "text", "text": "음수의 제곱근은 구할 수 없습니다."}], "is_error": True}
return {"content": [{"type": "text", "text": str(math.sqrt(args["n"]))}]}
calculator = create_sdk_mcp_server("calculator", tools=[add, multiply, sqrt])
한 프로젝트에서 인프로세스 서버와 외부 stdio 서버를 동시에 사용할 수 있습니다.
# 인프로세스: 앱 상태 접근
app_server = create_sdk_mcp_server("app", tools=[get_user, list_users])
options = ClaudeAgentOptions(
permission_mode="bypassPermissions",
mcp_servers={
# 인프로세스 서버
"app": app_server,
# 외부 stdio 서버
"postgres": {
"type": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres",
"postgresql://localhost:5432/mydb"],
},
},
allowed_tools=["get_user", "list_users"], # MCP 도구 이름으로 허용
)
ClaudeSDKClient를 사용하면 대화 중에 MCP 서버 상태를 조회하고 관리할 수 있습니다.
async with ClaudeSDKClient(options=options) as client:
# 서버 상태 확인
status = await client.get_mcp_status()
for server in status["mcpServers"]:
print(f"{server['name']}: {server['status']}")
if server["status"] == "connected" and "tools" in server:
for t in server["tools"]:
print(f" - {t['name']}: {t.get('description', '')}")
# 실패한 서버 재연결
for server in status["mcpServers"]:
if server["status"] == "failed":
print(f"{server['name']} 재연결 시도...")
await client.reconnect_mcp_server(server["name"])
# 서버 일시 비활성화
await client.toggle_mcp_server("heavy-server", enabled=False)
await client.query("가벼운 작업만 수행해줘")
async for msg in client.receive_response():
pass
# 다시 활성화
await client.toggle_mcp_server("heavy-server", enabled=True)
get_mcp_status()가 반환하는 각 서버의 상태 정보입니다.
class McpServerStatus(TypedDict):
name: str # 서버 이름
status: McpServerConnectionStatus # 연결 상태
serverInfo: McpServerInfo # 서버 정보 (연결 시)
error: str # 에러 메시지 (실패 시)
config: McpServerStatusConfig # 서버 설정
scope: str # 설정 범위
tools: list[McpToolInfo] # 도구 목록 (연결 시)
| status | 의미 | 대응 |
|---|---|---|
"connected" | 정상 | 도구 사용 가능 |
"pending" | 연결 대기 | 잠시 후 재확인 |
"failed" | 연결 실패 | reconnect_mcp_server() |
"needs-auth" | 인증 필요 | 토큰/크레덴셜 설정 |
"disabled" | 비활성화 | toggle_mcp_server(name, True) |
| 기준 | 인프로세스 (SDK) | 외부 서버 (stdio/SSE/HTTP) |
|---|---|---|
| 성능 | 빠름 (함수 호출) | 느림 (IPC/네트워크) |
| 앱 상태 접근 | 직접 접근 가능 | 불가 (별도 프로세스) |
| 배포 | 단일 프로세스 | 서버 별도 관리 |
| 언어 | Python만 | 아무 언어 |
| 생태계 | 직접 구현 | npm에 수백 개 서버 패키지 |
| 재사용 | 이 앱에서만 | 여러 앱에서 공유 |
판단 기준:
MCP 서버는 Claude Agent의 능력을 무한히 확장하는 핵심 기능입니다.
기억할 3가지:
@tool + create_sdk_mcp_server() — 3줄이면 커스텀 도구 완성allowed_tools에 MCP 도구 이름 추가 — 내장 도구와 동일하게 허용/차단 가능전체 소스 코드는 GitHub에서 확인할 수 있습니다:
👉 GitHub 레포지토리 링크