打造強大自主 AI agent代理系統的關鍵組合：LangGraph + MCP + Ollama

在本文中，將帶來一個精簡的實作教學，示範如何結合 LangGraph、MCP（Model Context Protocol）與 Ollama，快速構建一套多代理（multi-agent）聊天機器人系統，無論是應用於商業場景還是個人專案，皆能展現高度效能與彈性。

目前 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，並針對每筆工具調用進行以下步驟處理：

1. 抽取工具參數：
   從工具調用中解析 tool_name、tool_args 與 tool_id，支援 dict 與 object 兩種格式，以提升函式適應性與模組化程度。
2. 工具名稱匹配：
   根據 tool_name 對比當前可用工具清單，查找對應工具函式。若無法對應成功，則產生一則錯誤訊息，並列出所有有效工具名稱以供參考。
3. 函式類型判斷與執行：
   透過 asyncio.iscoroutinefunction() 判定工具是否為非同步函式。
4. • 若為 async，則以 await tool.coroutine(**tool_args) 執行。
   • 若為同步，則使用 tool.func(**tool_args) 或 tool(**tool_args) 執行。 此流程保證了同步與非同步工具可並存，並且能正確處理函數調用。
4. 錯誤處理與回報：
   所有執行過程皆以 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）中，有效提升多步驟交互與工具使用的靈活性與可維護性。

結構化設計概述：

1. 工具執行節點（"tools"）
   我將 async_tool_executor 函式封裝為一個名為 "tools" 的節點，並納入對話流程圖中。此節點專責處理 AI 所觸發的所有工具調用，具備動態辨識與執行多種同步／非同步工具的能力。
2. 動態路由器（Router Function）
   我實作了一個路由器函式，負責根據對話中最新一則訊息的內容決定對話的下一步邏輯。該函式會檢查是否存在 AI 所提出的 tool_calls：
3. • 若存在工具調用，流程將導向 "tools" 節點執行該操作。
   • 若無工具調用，則流程進入 "end" 節點，結束當前會話輪次。
3. 圖形流程建構（Graph Edges）
   為實現整體流程的順暢與可迭代，我設計了以下節點間的連接邏輯：
4. • 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 助手在執行過程中具備明確角色指引與即時語境感知。

功能設計說明：

1. 即時日期注入
   透過 datetime.now().strftime("%Y-%m-%d") 取得當前系統日期，並將其嵌入提示詞中。此設計可協助模型在生成內容時參考當日時間，有助於強化其「實時感知」能力，尤其在產出報告、時間敏感任務（如搜尋、摘要）時特別重要。
2. 助手角色與任務定義
   在提示詞中，我明確定義了 AI 助手的角色：具備分析、生成與執行能力的數位代理人，需根據使用者的請求進行判斷與回應，並在適當情況下主動呼叫工具支援完成任務。
3. 可用工具列舉
   系統提示詞中列出了代理可操作的三個主要工具，讓模型在推理階段能預期並選擇合適的工具執行指令：
4. • 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 回應。

設計重點：

1. 提示詞模板組合（Prompt Composition）
   利用 ChatPromptTemplate 技術，將系統指令與使用者訊息進行動態合併，確保每次呼叫 LLM（大語言模型）時皆包含：
2. • 當前上下文資訊（如：docs_info）
   • AI 助手的角色與行為規則（透過 get_system_prompt(docs_info) 動態生成）
2. 資料流管線化處理（LLM Pipelining）
   使用管線操作符 | 將格式化後的提示詞直接串接至 LLM，建立連貫且具模組彈性的處理流程。此設計風格可與 LangChain、LangGraph 等框架無縫對接。
3. 訊息標準化（Message Normalization）
   所有純文字輸入皆轉換為 HumanMessage 物件，維持訊息格式一致性，確保後續代理框架可正確解析訊息來源與意圖。
4. 訊息歷程維護（Conversation State Management）
   所有回應均以訊息物件方式附加於訊息串列中，使系統具備完整對話上下文追蹤能力，支援多輪查詢與狀態遞延推理。
5. 工具無關性（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