前天我們透過 FastMCP 成功建立一個簡單的 MCP Server，並提供 sayHello 工具來回傳打招呼訊息。
昨天我們則用 ADK (Agent Development Kit) 把 MCP Server 串進一個 Agent，讓 Agent 可以透過 MCP 工具完成打招呼。

今天，我們要挑戰的是改用 LangGraph 來搭建 Agent，把 MCP Server 以 stdio 的方式接上，讓 Agent 能在推論流程中呼叫 sayHello 工具。

1. MCP Server (回顧前天)

我們前天寫的 mcp_server.py 長這樣：

mcp = FastMCP("My MCP Server")

@mcp.tool
def sayHello(text: str) -> str:
    """回傳一段打招呼的訊息。"""
    return f"Hello, {text}!"

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

這個 Server 透過 mcp.run()，會使用 stdio 來跟外部溝通。

2. 在 LangGraph 端建立 Agent 串接 MCP Server

agent.py:

# 讀取環境變數
load_dotenv()

MODEL = os.getenv("OPENAI_API_MODEL", "gpt-4.1")

# === 系統提示 ===
SYSTEM_HINT = (
    "你是一個只會打招呼的助手。"
    "當使用者請你打招呼時，請呼叫 MCP 提供的 sayHello(text: str) 工具來產生訊息；"
    "若使用者需求與打招呼無關，直接回覆「我只能打招呼。」；"
    "所有回覆一律使用繁體中文。"
)

def _extract_text_from_result(result) -> str:
    try:
        if isinstance(result, dict) and "messages" in result and result["messages"]:
            last = result["messages"][-1]
            content = getattr(last, "content", None) or ""
            if isinstance(content, list):
                # 內容有可能是多段結構，取出文字段
                texts = []
                for part in content:
                    if isinstance(part, dict) and "text" in part:
                        texts.append(part["text"])
                    elif isinstance(part, str):
                        texts.append(part)
                return "\n".join([t for t in texts if t])
            if isinstance(content, str):
                return content
        return str(result)
    except Exception:
        return str(result)

async def run_agent(user_query: str):
    # 以 stdio 方式啟動 MCP server：python mcp_server.py
    params = StdioServerParameters(
        command="python",
        args=["mcp_server.py"],
        env=os.environ.copy(),
    )

    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await load_mcp_tools(session)

            llm = ChatOpenAI(model=MODEL)
            agent = create_react_agent(llm, tools)

            messages = [
                {"role": "system", "content": SYSTEM_HINT},
                {"role": "user", "content": user_query},
            ]

            result = await agent.ainvoke({"messages": messages})
            return _extract_text_from_result(result)

if __name__ == "__main__":
    out = asyncio.run(run_agent("幫我跟小明打招呼"))
    print(out)

3. 寫一個簡單的 Fastapi 端點進行測試

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class QueryRequest(BaseModel):
    query: str

@app.post("/query")
async def ask_agent(req: QueryRequest):
    text = await run_agent(req.query)
    return {"response": text}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

4. 透過 curl 直接進行測試

透過這幾天的練習，我們先用 FastMCP 建立了一個簡單的 MCP Server，接著嘗試用 ADK 與 LangGraph 把 MCP 工具串進 Agent，並且包裝成 FastAPI 服務。雖然我們的範例很簡單，只是「打招呼」，但這個模式可以輕鬆的去延伸到更複雜的應用場景，例如查資料庫、調用外部 API等等。只要 MCP 提供了對應的工具，Agent 就能在流程中進行推理與調用。