在本文中，將帶來一個精簡的實作教學，示範如何結合 LangGraph、MCP（Model Context Protocol）與 Ollama，快速構建一套多代理（multi-agent）聊天機器人系統，無論是應用於商業場景還是個人專案，皆能展現高度效能與彈性。
儘管 MCP 早在 2024 年 11 月就已正式推出，它直到近期才受到廣泛關注，並引發對其生命周期的討論。LangChain 團隊甚至在 X上發起了一場投票：投票結果顯示：40.8% 的參與者認為 MCP 是未來的標準協議；25.8% 則認為它只是曇花一現；其餘 33.4% 採觀望態度。
目前 MCP 已原生支援多種常見服務，例如 GitHub、Slack、PostgreSQL 等。由於它採用開放協議設計，可與各大主流 LLM（如 Claude、OpenAI、Gemini 等）無縫整合，這種跨平台兼容性令人驚豔。總結以進行式來看，MCP 能不能成為主流雖然還在未定之天，哪一天A2A橫空出世也未不可知。但我們還是得向前走，看看這個可能的明日之星如何應用。
實機展示：多代理 AI 聊天機器人如何運作
現在，我將示範一個線上聊天機器人，實際操作來說明整合效能。
第一個問題我會詢問：「你能寫一份有關最新大型語言模型（LLM）的報告嗎？」您也可以隨時發問其他問題。
觀察這個機器人產出內容的過程，可以看出它透過 create_chatbot 函數所驅動的結構化流程，結合系統指令、使用者輸入與工具執行機制，有效進行任務處理。系統會依據查詢內容，智能選擇合適的工具。
針對上述問題，它會呼叫 Google 搜尋工具以擷取最新資料，再生成報告。當我問出第二個問題：「請寫一個使用 Seaborn 繪製帶有迴歸線的散佈圖的 Python 程式」，系統則會根據查詢類型，自動導向如 python_repl 或 data_visualization 等適合的模組。async_tool_executor 負責處理這些工具呼叫，無論同步或非同步任務，如程式碼生成或資料視覺化，都能被穩定執行。
整個對話過程中，StateGraph 模組負責管理對話狀態，確保機器人能維持上下文一致性並準確回應後續問題；get_tools 函數確保僅提供功能正常的工具，以維持系統穩定；而主函數則整合輸入、工具調用與輸出處理，使整體流程即時運作、無縫協同。
透過這個展示，您將能清楚分辨 MCP 與傳統 Function Call 的差異、使用場景的取捨原則，以及如何實際運用 LangGraph、MCP 與開源技術打造一個強大且具擴展性的 AI 代理聊天系統。
MCP 與 Function Call 的差異為何？
Function Call 機制下的 AI 像是一位根據劇本精確執行任務的技術人員——它能呼叫特定工具並執行既定流程，但限制在固定範圍內、必須依序進行。
相對地，MCP 提供更高自由度，讓 AI 像是一位擁有工具箱的智慧代理，它不僅能探索並選擇工具，還能組合並執行多步驟任務，展現出更強的自我調節能力。
Function Call 與模型提示緊密耦合，適合控制嚴謹、流程簡單的任務，但彈性相對有限。MCP 則透過開放協議實現鬆散耦合，具有極高彈性與擴充性，唯需更謹慎的系統設計來控管流程複雜性與安全性。
什麼情況下該選擇 Function Call 或 MCP？
選擇 Function Call 適用於以下場景：
- 任務結構明確、動作單一的情況（如表單填寫、數據查詢等）。
- 須產出高度結構化的輸出內容。
- 專案整合需求簡單、偏好輕量級解決方案。
選擇 MCP 則適用於以下情境：
- 任務具有多步驟流程，或需要根據上下文動態調整。
- 系統需與多個異質資料源整合，如企業內部系統。
- AI 需要長時間記憶上下文並展現決策彈性。
值得注意的是，Function Call 與 MCP 並非互斥關係——實務上可互為補充。例如，在 MCP 架構中可內嵌 Function Call，用於處理模型返回的結構化輸出。
從設計理念來看，Function Call 是一種自然語言轉函數執行的受控方式；而 MCP 則是一個讓 AI 能自由探索與操作執行環境的廣域通訊介面。
實作開始：建構 MCP 應用的第一步
現在我們將實際動手操作，逐步說明如何建立一個基於 MCP 架構的應用。
首先，匯入核心函式庫，這些元件在後續開發中將扮演關鍵角色：
langchain_mcp_adapters：將 MCP 工具轉換為 LangChain 可用格式，方便與 LangGraph 整合，並提供多 MCP 伺服器串接能力。MCP：一種開放協議，用以標準化應用程式如何提供上下文給大型語言模型。googlesearch-python：一個簡易封裝的 Google 搜尋工具，可快速整合查詢功能。
# agent.py
from langchain_core.messages import AIMessage, ToolMessage, HumanMessage
from langgraph.graph import StateGraph, START, END, MessagesState
from nodes import create_chatbot
import asyncio
import os
import dotenv
from langchain_mcp_adapters.client import MultiServerMCPClient
# main.py
import streamlit as st
import asyncio
from agent import create_agent
from langchain_core.messages import HumanMessage
# nodes.py
from server import get_tools
from langgraph.graph import MessagesState
from langchain_openai import ChatOpenAI
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from datetime import datetime
import os
# server.py - unchanged except for removing search_google if it's not working
from mcp.server.fastmcp import FastMCP
from langchain_experimental.utilities import PythonREPL
import io
import base64
import matplotlib.pyplot as plt
from openai import OpenAI
from pydantic import BaseModel, Field
import os
from dotenv import load_dotenv
import asyncio
from googlesearch import search
agent.py
在 Agent.py 中，我設計了一個名為 create_agent 的非同步函式，作為建構 AI 代理的主要進入點。該函式支援一個可選參數 docs_info，用於向聊天機器人注入相關的文件資訊，以增強其回應的語境精準度。
為實現與 MCP 伺服器的高效通訊，我整合了 MultiServerMCPClient，並以 async with 的上下文管理器方式封裝連線過程。該客戶端連接至本地 MCP 伺服器 (http://localhost:8000/sse)，並透過 Server-Sent Events（SSE）協議進行事件串流，設置連線超時為 30 秒，確保在保持即時性的同時控制連線壽命。
在與 MCP 伺服器建立連線後，我透過 client.get_tools() 呼叫以動態擷取可用的 MCP 工具。這些工具支援進階功能，如程式碼執行、資料查詢與視覺化等，有助於擴展聊天機器人的應用場景。
為管理對話過程中的上下文與狀態流轉，我使用 MessagesState 建構了一個 StateGraph，此圖形狀態機為多輪對話提供結構化狀態管理機制。
最後，我透過 create_chatbot(docs_info) 函式建立了一個聊天機器人節點，該節點不僅能處理來自使用者的查詢，還能有效整合提供的文檔資訊進行推理與互動，實現強大的知識驅動型 AI 對話體驗。
async def create_agent(docs_info=None):
async with MultiServerMCPClient(
{
"server":{
"url":"http://localhost:8000/sse",
"transport":"sse",
"timeout": 30
}
}
) as client:
# Get MCP tools
tools = client.get_tools()
# Create the graph builder
graph_builder = StateGraph(MessagesState)
# Create nodes
chatbot_node = create_chatbot(docs_info)
graph_builder.add_node("chatbot", chatbot_node)
在本模組中，我設計了一個名為 async_tool_executor 的非同步函式，負責動態解析並執行 AI 代理中的工具調用邏輯。該函式接收 state 作為輸入，其中包含對話歷程中的所有訊息列表（messages），作為代理決策的上下文依據。
首先，我從 messages[-1] 擷取最新訊息，檢查是否包含工具調用（tool calls）。這些調用可能直接出現在訊息中 (tool_calls)，或隱藏於 additional_kwargs 屬性內，藉此兼容多種代理框架的訊息格式。
若該訊息未觸發任何工具調用，則直接返回原始 messages，無需進一步處理。
當偵測到工具調用後，我會將原訊息複製至 new_messages，並針對每筆工具調用進行以下步驟處理：
- 抽取工具參數：
從工具調用中解析tool_name、tool_args與tool_id，支援 dict 與 object 兩種格式，以提升函式適應性與模組化程度。 - 工具名稱匹配：
根據tool_name對比當前可用工具清單，查找對應工具函式。若無法對應成功，則產生一則錯誤訊息，並列出所有有效工具名稱以供參考。 - 函式類型判斷與執行：
透過asyncio.iscoroutinefunction()判定工具是否為非同步函式。 - 若為 async，則以 await tool.coroutine(**tool_args) 執行。
- 若為同步，則使用 tool.func(**tool_args) 或 tool(**tool_args) 執行。 此流程保證了同步與非同步工具可並存，並且能正確處理函數調用。
- 錯誤處理與回報：
所有執行過程皆以 try-except 包裝，當工具執行時出現例外錯誤時，系統會將錯誤詳情包裝成訊息附加至new_messages，提供使用者與開發者透明回饋。
整體設計具備下列優勢：
- 高度動態性：可根據對話內容自動調度多工具執行，無需硬編碼。
- 非同步支援：充分利用 Python 的 async/await 特性，加速工具響應效率。
- 錯誤韌性強：支援逐步降級處理與細緻錯誤回報。
- 格式彈性高：兼容多種輸入格式與框架輸出，易於擴充與整合。
# Custom async tool node to handle async MCP tools
async def async_tool_executor(state):
messages = state["messages"]
last_message = messages[-1]
# Check if there are tool calls
tool_calls = None
if hasattr(last_message, "tool_calls"):
tool_calls = last_message.tool_calls
elif hasattr(last_message, "additional_kwargs") and "tool_calls" in last_message.additional_kwargs:
tool_calls = last_message.additional_kwargs["tool_calls"]
if not tool_calls:
return {"messages": messages}
# Process each tool call
new_messages = messages.copy()
for tool_call in tool_calls:
# Handle different formats of tool_call
if isinstance(tool_call, dict):
tool_name = tool_call.get("name")
tool_args = tool_call.get("args", {})
tool_id = tool_call.get("id", "tool-call-id")
else:
tool_name = tool_call.name
tool_args = tool_call.args if hasattr(tool_call, "args") else {}
tool_id = getattr(tool_call, "id", "tool-call-id")
# Print debug info
print(f"Executing tool: {tool_name}")
print(f"Tool args: {tool_args}")
# Find the matching tool
tool = next((t for t in tools if t.name == tool_name), None)
if not tool:
# Tool not found
tool_error = f"Error: {tool_name} is not a valid tool, try one of {[t.name for t in tools]}."
new_messages.append(AIMessage(content=tool_error))
else:
try:
# Execute the async tool
if asyncio.iscoroutinefunction(tool.coroutine):
result = await tool.coroutine(**tool_args)
else:
# Fall back to sync execution if needed
result = tool.func(**tool_args) if hasattr(tool, 'func') else tool(**tool_args)
print(f"Tool result: {result}")
# Add tool result
new_messages.append(ToolMessage(
content=str(result),
tool_call_id=tool_id,
name=tool_name
))
except Exception as e:
# Handle errors
error_msg = f"Error: {str(e)}\n Please fix your mistakes."
print(f"Tool error: {error_msg}")
new_messages.append(AIMessage(content=error_msg))
return {"messages": new_messages}
在本系統中，我設計了一個具備結構化對話流程的聊天代理，透過將非同步工具執行節點與動態路由判斷整合至對話圖（StateGraph）中，有效提升多步驟交互與工具使用的靈活性與可維護性。
結構化設計概述：
- 工具執行節點（"tools"）
我將async_tool_executor函式封裝為一個名為"tools"的節點，並納入對話流程圖中。此節點專責處理 AI 所觸發的所有工具調用，具備動態辨識與執行多種同步／非同步工具的能力。 - 動態路由器（Router Function）
我實作了一個路由器函式，負責根據對話中最新一則訊息的內容決定對話的下一步邏輯。該函式會檢查是否存在 AI 所提出的tool_calls： - 若存在工具調用，流程將導向 "tools" 節點執行該操作。
- 若無工具調用，則流程進入 "end" 節點，結束當前會話輪次。
- 圖形流程建構（Graph Edges）
為實現整體流程的順暢與可迭代，我設計了以下節點間的連接邏輯： - chatbot → 依據 router 函式結果，條件導向 tools 或 end
- tools → chatbot：形成迴圈流程，允許 AI 代理在工具操作後返回對話主流程，持續互動與回應
整體效益：
- 可擴充性：此架構可輕鬆擴展至多工具、多流程情境，並允許根據訊息動態改變路由決策。
- 模組化維護：每個節點獨立實作，方便日後新增功能或修改行為邏輯。
- 多輪互動支持：
tools → chatbot的迴圈設計允許代理根據多次工具回應逐步建構答案，支援更複雜的任務處理。
# Add the async tool executor node
graph_builder.add_node("tools", async_tool_executor)
# Define router function to handle tool calls
def router(state):
messages = state["messages"]
last_message = messages[-1]
has_tool_calls = False
if isinstance(last_message, AIMessage):
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
has_tool_calls = True
elif hasattr(last_message, "additional_kwargs") and last_message.additional_kwargs.get("tool_calls"):
has_tool_calls = True
return "tools" if has_tool_calls else "end"
# Add edges
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges(
"chatbot",
router,
{
"tools": "tools",
"end": END
}
)
graph_builder.add_edge("tools", "chatbot")
# Compile the graph
graph = graph_builder.compile()
return graph, client # Return client to keep it alive
node.py
緊接著在 Node.py 中，我實作了一個名為 get_system_prompt 的函式，用以動態生成系統提示詞（system prompt），確保 AI 助手在執行過程中具備明確角色指引與即時語境感知。
功能設計說明：
- 即時日期注入
透過datetime.now().strftime("%Y-%m-%d")取得當前系統日期，並將其嵌入提示詞中。此設計可協助模型在生成內容時參考當日時間，有助於強化其「實時感知」能力，尤其在產出報告、時間敏感任務（如搜尋、摘要）時特別重要。 - 助手角色與任務定義
在提示詞中，我明確定義了 AI 助手的角色：具備分析、生成與執行能力的數位代理人，需根據使用者的請求進行判斷與回應，並在適當情況下主動呼叫工具支援完成任務。 - 可用工具列舉
系統提示詞中列出了代理可操作的三個主要工具，讓模型在推理階段能預期並選擇合適的工具執行指令： - generate_image：基於 DALL·E 的圖像生成工具，可根據文本創造視覺內容。
- data_visualization：以 matplotlib 為基底的資料視覺化工具，支援圖表繪製。
- python_repl：Python 執行環境，允許代理執行任意 Python 程式碼，適合數據分析、數值計算與資料處理。
設計優勢：
- 上下文穩定性：系統提示詞明確設定模型邊界與能力範圍，有助於穩定模型行為。
- 工具導向思維：引導模型主動思考何時使用工具而非僅以文字回應，促進 agentic pattern 的落實。
- 模組化可擴充：未來可輕鬆加入更多工具或根據使用場景動態更換提示內容。
def get_system_prompt(docs_info=None):
system_prompt = f"""
Today is {datetime.now().strftime("%Y-%m-%d")}
You are a helpful AI Assistant that can use these tools:
- generate_image: Generate an image using DALL-E based on a prompt
- data_visualization: Create charts with Python and matplotlib
- python_repl: Execute Python code
When you call image generation or data visualization tool, only answer the fact that you generated, not base64 code or url.
Once you generated image by a tool, then do not call it again in one answer.
"""
if docs_info:
docs_context = "\n\nYou have access to these documents:\n"
for doc in docs_info:
docs_context += f"- {doc['name']}: {doc['type']}\n"
system_prompt += docs_context
system_prompt += "\nYou should always answer in same language as user's ask."
return system_prompt
在本模組中，我設計並實作了 create_chatbot 函式，作為整體代理對話流程的核心處理引擎。此函式專責接收使用者輸入，結合系統提示語（System Prompt）與語言模型推理結果，生成具上下文與邏輯一致性的 AI 回應。
設計重點：
- 提示詞模板組合（Prompt Composition）
利用ChatPromptTemplate技術，將系統指令與使用者訊息進行動態合併，確保每次呼叫 LLM（大語言模型）時皆包含： - 當前上下文資訊（如：docs_info）
- AI 助手的角色與行為規則（透過 get_system_prompt(docs_info) 動態生成）
- 資料流管線化處理（LLM Pipelining）
使用管線操作符|將格式化後的提示詞直接串接至 LLM，建立連貫且具模組彈性的處理流程。此設計風格可與 LangChain、LangGraph 等框架無縫對接。 - 訊息標準化（Message Normalization）
所有純文字輸入皆轉換為HumanMessage物件，維持訊息格式一致性，確保後續代理框架可正確解析訊息來源與意圖。 - 訊息歷程維護（Conversation State Management）
所有回應均以訊息物件方式附加於訊息串列中，使系統具備完整對話上下文追蹤能力，支援多輪查詢與狀態遞延推理。 - 工具無關性（Tool-Agnostic Design）
該函式不綁定特定工具執行邏輯，使其在 LangGraph 流程中可彈性搭配各種工具執行節點。當需要工具輔助時，系統將自動透過外部節點處理（如 async_tool_executor），實現高模組化、高解耦的設計理念。
功能優勢：
- ✅ 靈活性高：可根據任務動態調整 prompt 與訊息邏輯
- ✅ 上下文敏感：可處理長序列、多步驟互動情境
- ✅ 高可維護性：模組分離明確，便於日後升級或替換模型
- ✅ 強互動性：完整訊息歷程與系統指令機制，支援主動式 AI 行為
def create_chatbot(docs_info=None):
prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(get_system_prompt(docs_info)),
HumanMessagePromptTemplate.from_template("{input}")
])
# Use the LLM without binding tools
chain = prompt | llm
def chatbot(state: MessagesState):
# Ensure messages are in the right format
if isinstance(state["messages"], str):
from langchain_core.messages import HumanMessage
messages = [HumanMessage(content=state["messages"])]
else:
messages = state["messages"]
response = chain.invoke(messages)
return {"messages": messages + [response]}
return chatbot
在 Server.py 中，我設計並實作了一組可模組化的 AI 工具（Tools），以支援代理系統在自然語言互動中進行多種任務處理，包含圖像生成、資料視覺化以及 Python 程式碼即時執行。這些工具皆可獨立運作，並支援非同步調用，符合 MCP 與 LangGraph 架構下的動態代理運作需求。
🔧 工具設計詳述
1. generate_image: 圖像生成工具（基於 DALL·E）
- 功能：根據使用者提供的自然語言提示（prompt）產生圖片。
- 實作邏輯：
- 驗證輸入提示是否為合法字串。
- 透過非同步方式調用 OpenAI API（DALL·E）。
- 成功時回傳圖片 URL，失敗時提供錯誤回饋。
- 應用場景：用於創意生成、行銷素材提案、設計草圖起稿等。
2. data_visualization: 資料視覺化工具（基於 matplotlib）
- 功能：執行使用者提供的資料視覺化 Python 程式碼。
- 實作細節：
- 執行使用者腳本以生成圖表。
- 圖表輸出為 PNG 格式，並以 base64 編碼回傳，便於在前端或 UI 中即時呈現。
- 應用場景：適用於數據探索、分析簡報、圖形報告產出等場景。
3. python_repl: Python 程式碼執行環境（REPL）
- 功能：提供可即時執行任意 Python 程式碼的 REPL（Read-Eval-Print Loop）介面。
- 處理邏輯：
- 安全地解析與執行輸入的程式碼。
- 捕捉執行錯誤，並以清晰訊息回饋使用者。
- 應用場景：快速算法驗證、數據處理、自訂邏輯測試等。
✅ 系統特性與優勢
- 非同步支援：所有工具皆可透過
async方法進行整合，提升代理整體執行效率。 - 錯誤處理強化：每個工具皆內建例外處理機制，能即時回報問題並維持對話流暢。
- 介面一致性：所有工具皆依據 MCP 標準設計，確保與 LangGraph、LangChain 等代理框架可高度整合。
- 可擴充性強：未來可輕鬆加入其他工具（如 Web 搜尋、SQL 查詢等），進一步擴展代理能力。
@mcp.tool()
async def generate_image(prompt: str) -> str:
"""
Generate an image using DALL-E based on the given prompt.
"""
if not isinstance(prompt, str) or not prompt.strip():
raise ValueError("Invalid prompt")
try:
# Since this is an async function, we need to handle the synchronous OpenAI call properly
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: client.images.generate(
model="dall-e-3",
prompt=prompt,
size="1024x1024",
quality="standard",
n=1
)
)
# Return both success message and URL
return f"Successfully generated an image of {prompt}! Here's the URL: {response.data[0].url}"
except Exception as e:
return f"Error generating image: {str(e)}"
repl = PythonREPL()
@mcp.tool()
def data_visualization(code: str):
"""Execute Python code. Use matplotlib for visualization."""
try:
repl.run(code)
buf = io.BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
img_str = base64.b64encode(buf.getvalue()).decode()
return f"data:image/png;base64,{img_str}"
except Exception as e:
return f"Error creating chart: {str(e)}"
@mcp.tool()
def python_repl(code: str):
"""Execute Python code."""
return repl.run(code)
在本模組中，我實作了 get_tools 函式與 MCP 伺服器啟動邏輯，確保代理系統在執行階段僅載入已註冊且功能正常的工具，並透過 SSE（Server-Sent Events）進行即時通訊。
🔧 get_tools: 工具列表生成器
此函式負責回傳目前可供代理系統使用的工具集合。設計上具備擴充性與防錯能力：
- 預設工具集：
- generate_image：基於 DALL·E 的圖像生成器。
- python_repl：支援任意 Python 程式碼執行的 REPL 環境。
- data_visualization：資料視覺化工具，內建 matplotlib 圖表輸出。
- 動態擴充能力：
- 若外部傳入 retriever_tool（如文檔檢索器、向量搜尋引擎等），將其納入工具集，實現條件式註冊。
- 回傳格式：
- 輸出為標準化的工具物件陣列，符合 MCP / LangGraph 的工具定義協議。
此函式的設計確保代理系統僅啟用功能完整且必要的工具，提升整體系統穩定性與安全性。
🚀 MCP 系統啟動：非同步事件驅動通訊
在腳本執行階段，我透過以下語句啟動 MCP Server：
mcp.run(transport="sse")
- 傳輸協議：
sse（Server-Sent Events） - 採用事件流推送模型，適用於即時雙向通訊場景（例如工具回傳進度、即時回應生成等）。
- 與 Web 瀏覽器或代理框架具備高度相容性。
- 啟動結果：
- 工具註冊完成後，MCP Server 將進入待命狀態，準備接受 LangGraph 或其他代理架構的呼叫。
- 所有工具功能皆可透過標準協議暴露給多代理系統進行交互使用。
✅ 優化特點
- 模組化註冊機制：便於工具維護與測試，可針對特定場景做動態載入。
- 事件驅動通訊：降低延遲，增強代理互動的即時性與回應流暢度。
- 與 LangGraph 無縫整合：可直接嵌入 LangGraph 節點流程，支援結構化對話與多階段任務處理。
def get_tools(retriever_tool=None):
# Only include tools that are working
base_tools = [generate_image, python_repl, data_visualization]
if retriever_tool:
base_tools.append(retriever_tool)
return base_tools
if __name__ == "__main__":
mcp.run(transport="sse")
main.py在 main.py 中，我設計並實作了 main() 函式，作為整體代理系統的非同步執行入口。此函式負責建立代理實例、處理使用者輸入、調用對應工具，並以即時互動方式回應使用者，實現完整的多工具 AI 聊天代理運行流程。
🔄 main() 函式功能概述
1. 初始化代理與客戶端
agent, client = await create_agent()
- 使用
create_agent()非同步方法初始化代理流程與 MCP 客戶端連線。 - 建立後，系統具備處理多工具任務的能力，並可即時調用遠端工具資源。
2. 接收使用者輸入
user_input = input("Your query: ")
message = HumanMessage(content=user_input)
- 透過 CLI 介面接收使用者問題，並轉換為標準格式的
HumanMessage，便於後續代理系統解析與處理。
3. 非同步代理呼叫與訊息處理
response = await agent.ainvoke(message)
- 非同步執行代理回應流程，支援跨工具、多步驟的任務處理。
- 回傳訊息會依據來源進行分類顯示：
- 使用者訊息（UserMessage）：印出原始輸入內容。
- 工具回應（ToolMessage）：印出執行結果；若包含圖片生成（如來自 generate_image 工具），則進一步解析並顯示圖片 URL。
4. 錯誤處理
except Exception as e:
print(f"Error: {str(e)}")
- 系統內建例外處理機制，能攔截代理過程中可能發生的錯誤，並以清晰訊息提示開發者或使用者。
5. 維持客戶端存活
await client.ainvoke()
- 持續保持 MCP 連線，確保工具與代理之間能穩定通信，尤其在多步驟任務處理中不可中斷。
6. 非同步執行入口
asyncio.run(main())
- 使用 Python 3.7+ 標準的
asyncio事件迴圈執行主流程，確保所有非同步邏輯可完整運作。
✅ 系統特性與優勢
- 完整事件驅動架構：支援多工具並行處理與狀態管理。
- 模組化設計：可快速替換、擴充工具與代理策略。
- 即時互動體驗：適用於 CLI 工具、測試場景與開發者原型測試。
- 高容錯性：強化錯誤處理與回應訊息提示，適用於開發環境與實際部署。
async def main():
# Create the agent
agent, client = await create_agent()
# Get user input from command line
user_input = input("What would you like to ask? ")
# Create a proper initial message
initial_message = HumanMessage(content=user_input)
try:
# Use the agent asynchronously
print("Processing your request...")
result = await agent.ainvoke({"messages": [initial_message]})
# Print the results
for message in result["messages"]:
if hasattr(message, "type") and message.type == "human":
print(f"User: {message.content}")
elif hasattr(message, "type") and message.type == "tool":
print(f"Tool Result: {message.content}")
# If it's an image generation result, extract URL
if "image" in message.content.lower() and "url" in message.content.lower():
print("Image Generated Successfully!")
else:
print(f"AI: {message.content}")
except Exception as e:
print(f"Error: {str(e)}")
# Keep the client alive until all operations are done
# In a real application, you'd keep the client active as long as needed
if __name__ == "__main__":
asyncio.run(main())
結語：
MCP（Model Context Protocol）提供了一套開放且通用的標準，讓 AI 系統能夠以一致的方式整合來自各種資料來源、工具與服務。這大幅減少了針對每一個資料源建立專屬連接器的需求，從而簡化了 AI 系統的整合與部署流程。
透過 MCP，即使是目前尚未具備智慧功能的傳統服務，也能將自身的功能以「工具」的形式對外公開，供大型語言模型（LLM）呼叫與操作。這使得 LLM 可以無縫地與現有系統互動、執行任務，而無需對既有架構進行重大改動。
https://github.com/modelcontextprotocol
https://modelcontextprotocol.io/introduction
