AI 에이전트 개발 실전 #6 MCP 서버 직접 만들기

pip install "mcp[cli]"

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("order-tools")

@mcp.tool()
def search_orders(query: str) -> str:
    """주문 번호, 고객 이름, 날짜로 주문을 검색한다.
    사용자가 특정 주문의 상태나 내역을 물으면 먼저 이 도구로 주문을 찾는다."""
    results = do_search(query)          # 실제 구현
    return format_results(results)

@mcp.tool()
def get_order_detail(order_id: str) -> str:
    """주문 번호로 주문 상세를 조회한다. 예: ORD-2026-0001"""
    return fetch_detail(order_id)

if __name__ == "__main__":
    mcp.run()   # 기본은 stdio 방식으로 실행된다

import asyncio
import anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

client = anthropic.AsyncAnthropic()

async def run_agent(goal: str, max_steps: int = 20) -> str:
    server = StdioServerParameters(command="python", args=["order_server.py"])
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as mcp_session:
            await mcp_session.initialize()

            # 1. 서버의 도구 목록을 Claude tools 형식으로 변환
            listed = await mcp_session.list_tools()
            tools = [
                {
                    "name": t.name,
                    "description": t.description,
                    "input_schema": t.inputSchema,
                }
                for t in listed.tools
            ]

            # 2. 루프는 1편과 같다. 도구 실행만 MCP 호출로 바뀐다.
            messages = [{"role": "user", "content": goal}]
            for _ in range(max_steps):
                response = await client.messages.create(
                    model="claude-opus-4-8",
                    max_tokens=16000,
                    tools=tools,
                    messages=messages,
                )
                if response.stop_reason == "end_turn":
                    return next(b.text for b in response.content if b.type == "text")

                messages.append({"role": "assistant", "content": response.content})
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        result = await mcp_session.call_tool(block.name, block.input)
                        text = "\n".join(c.text for c in result.content if c.type == "text")
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": text,
                            "is_error": bool(result.isError),
                        })
                messages.append({"role": "user", "content": tool_results})

    raise RuntimeError("단계 한도 안에 작업을 끝내지 못했습니다.")

print(asyncio.run(run_agent("ORD-2026-0001 주문 상태를 알려줘.")))