LangGraph × FastMCP 2.0 強(qiáng)強(qiáng)聯(lián)合：輕松構(gòu)建企業(yè)級(jí)AI工作流，效率翻倍！原創(chuàng)

AI博物院

發(fā)布于 2025-9-8 07:00

瀏覽

1收藏

LangGraph 與 FastMCP 2.0（MCP 2.0）的結(jié)合，主要有兩種典型方式：

客戶端集成：將 FastMCP 服務(wù)器暴露的工具、提示與資源，作為 LangGraph 工作流中的可調(diào)用節(jié)點(diǎn)（最常見(jiàn)）。
服務(wù)端封裝：把已有的 LangGraph Agent 封裝為 FastMCP 服務(wù)器，對(duì)外以標(biāo)準(zhǔn)協(xié)議提供能力（反向集成）。

客戶端集成

MCP 模擬服務(wù)

用于本地測(cè)試與調(diào)試的 MCP 2.0 模擬服務(wù)。

以下示例基于 FastMCP 2.0 編寫(xiě)。

uv add fastmcp

示例代碼

最基礎(chǔ)的算術(shù)工具：

from fastmcp import FastMCP

mcp = FastMCP("MyServer")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    # Start an HTTP server on port 8000
    mcp.run(transport="http", host="127.0.0.1", port=8000)
uv run concepts-mcp/mcp_server_sample.py

運(yùn)行后，控制臺(tái)將輸出：

╭────────────────────────────────────────────────────────────────────────────╮
│                                                                            │
│        _ __ ___  _____           __  __  _____________    ____    ____     │
│       _ __ ___ .'____/___ ______/ /_/  |/  / ____/ __ \  |___ \  / __ \    │
│      _ __ ___ / /_  / __ `/ ___/ __/ /|_/ / /   / /_/ /  ___/ / / / / /    │
│     _ __ ___ / __/ / /_/ (__  ) /_/ /  / / /___/ ____/  /  __/_/ /_/ /     │
│    _ __ ___ /_/    \____/____/\__/_/  /_/\____/_/      /_____(*)____/      │
│                                                                            │
│                                                                            │
│                                FastMCP  2.0                                │
│                                                                            │
│                                                                            │
│               ???  Server name:     MyServer                                 │
│               ?? Transport:       Streamable-HTTP                          │
│               ?? Server URL:      http://127.0.0.1:8000/mcp                │
│                                                                            │
│               ???  FastMCP version: 2.12.0                                   │
│               ?? MCP SDK version: 1.13.1                                   │
│                                                                            │
│               ?? Docs:            https://gofastmcp.com                    │
│               ?? Deploy:          https://fastmcp.cloud                    │
│                                                                            │
╰────────────────────────────────────────────────────────────────────────────╯


[09/01/25 22:54:03] INFO     Starting MCP server 'MyServer' with transport 'http' on                      server.py:1571
                             http://127.0.0.1:8000/mcp
INFO:     Started server process [24590]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

驗(yàn)證功能

安裝并啟動(dòng) MCP Inspector 使用 npm 安裝 MCP Inspector：

npm install -g @modelcontextprotocol/inspector

或者，你可以直接使用npx運(yùn)行Inspector，無(wú)需全局安裝：

npx @modelcontextprotocol/inspector

啟動(dòng)后，Inspector 將在瀏覽器中打開(kāi)，默認(rèn)地址為 http://127.0.0.1:6274。

npx @modelcontextprotocol/inspector
Starting MCP inspector...
?? Proxy server listening on 127.0.0.1:6277
?? Session token: ab9f8a6edafccaa8a276b5bfebfc1d4t0fe486b14b5b42208d11d777cd7f17b4
Use this token to authenticate requests or set DANGEROUSLY_OMIT_AUTH=true to disable auth

?? Open inspector with token pre-filled:
   http://localhost:6274/?MCP_PROXY_AUTH_TOKEN=ab9f8a6edafccaa8a276b5bfebfc1d4t0fe486b14b5b42208d11d777cd7f17b4

?? MCP Inspector is up and running at http://127.0.0.1:6274 ??

模擬服務(wù)啟動(dòng)后，Inspector 會(huì)自動(dòng)發(fā)現(xiàn)并展示 MyServer 服務(wù)器的全部工具。

LangGraph × FastMCP 2.0 強(qiáng)強(qiáng)聯(lián)合：輕松構(gòu)建企業(yè)級(jí)AI工作流，效率翻倍！-AI.x社區(qū)

MCP Inspector

用于快速驗(yàn)證工具可用性，避免集成過(guò)程中的干擾項(xiàng)。

LangGraph 調(diào)用 MCP 服務(wù)

模型上下文協(xié)議（MCP）是一項(xiàng)開(kāi)放標(biāo)準(zhǔn)，用于規(guī)范應(yīng)用如何向語(yǔ)言模型提供工具與上下文。借助 langchain-mcp-adapters，LangGraph 智能體可直接使用 MCP 服務(wù)器上定義的工具。

LangGraph × FastMCP 2.0 強(qiáng)強(qiáng)聯(lián)合：輕松構(gòu)建企業(yè)級(jí)AI工作流，效率翻倍！-AI.x社區(qū)

BAML

模型上下文協(xié)議 (MCP) （圖片源自 LangGraph ）

安裝 langchain-mcp-adapters，使 LangGraph 能調(diào)用 MCP 工具。

uv add langchain-mcp-adapters
uv add langchain langchain-openai langchain-deepseek langgraph python-dotenv

工程根目錄添加 ??.env?? 文件

DEEPSEEK_API_KEY=sk-……

編寫(xiě)代碼 langgraph_use_mcp_as_client.py：

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode

import os
from dotenv import load_dotenv

# 加載.env文件中的環(huán)境變量
load_dotenv()

# Initialize the model
model = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

# Set up MCP client
client = MultiServerMCPClient(
    {
        "math": {
            # make sure you start your math server on port 8000
            "url": "http://127.0.0.1:8000/mcp/",
            "transport": "streamable_http",
        }
    }
)

asyncdef main():
    # Get tools from MCP server
    print("\n=== 獲取MCP工具 ===")
    tools = await client.get_tools()
    print(f"可用工具: {[tool.name for tool in tools]}")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}")
    
    # Bind tools to model
    print("\n=== 綁定工具到模型 ===")
    model_with_tools = model.bind_tools(tools)
    print(f"已將 {len(tools)} 個(gè)工具綁定到模型")
    
    # Create ToolNode
    tool_node = ToolNode(tools)
    
    def should_continue(state: MessagesState):
        messages = state["messages"]
        last_message = messages[-1]
        if last_message.tool_calls:
            return"tools"
        return END
    
    # Define call_model function
    asyncdef call_model(state: MessagesState):
        messages = state["messages"]
        print("\n=== 調(diào)用LLM模型 ===")
        print(f"輸入消息數(shù)量: {len(messages)}")
        if messages:
            print(f"最新消息: {messages[-1].content if hasattr(messages[-1], 'content') else str(messages[-1])}")
        
        response = await model_with_tools.ainvoke(messages)
        print(f"模型響應(yīng)類(lèi)型: {type(response).__name__}")
        if hasattr(response, 'content'):
            print(f"響應(yīng)內(nèi)容: {response.content}")
        if hasattr(response, 'tool_calls') and response.tool_calls:
            print(f"工具調(diào)用: {len(response.tool_calls)} 個(gè)")
            for i, tool_call in enumerate(response.tool_calls):
                print(f"  工具 {i+1}: {tool_call['name']} - 參數(shù): {tool_call['args']}")
        
        return {"messages": [response]}
    
    # Build the graph
    print("\n=== 構(gòu)建LangGraph工作流 ===")
    builder = StateGraph(MessagesState)
    builder.add_node("call_model", call_model)
    builder.add_node("tools", tool_node)
    print("已添加節(jié)點(diǎn): call_model (模型調(diào)用) 和 tools (工具執(zhí)行)")
    
    builder.add_edge(START, "call_model")
    builder.add_conditional_edges(
        "call_model",
        should_continue,
    )
    builder.add_edge("tools", "call_model")
    
    # Compile the graph
    graph = builder.compile()
    
    # Test the graph
    print("\n=== 開(kāi)始測(cè)試數(shù)學(xué)計(jì)算 ===")
    test_question = "what's (3 + 5) x 12?"
    print(f"測(cè)試問(wèn)題: {test_question}")
    
    math_response = await graph.ainvoke(
        {"messages": [{"role": "user", "content": test_question}]}
    )
    
    print("\n=== 最終結(jié)果 ===")
    print(f"消息鏈長(zhǎng)度: {len(math_response['messages'])}")
    for i, msg in enumerate(math_response['messages']):
        msg_type = type(msg).__name__
        if hasattr(msg, 'content'):
            print(f"消息 {i+1} ({msg_type}): {msg.content}")
        else:
            print(f"消息 {i+1} ({msg_type}): {str(msg)}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

測(cè)試結(jié)果

MCP 服務(wù)器：已在 http://127.0.0.1:8000/mcp/ 成功運(yùn)行，提供 add 與 multiply 工具。

MCP 客戶端：已成功連接到服務(wù)器并完成數(shù)學(xué)計(jì)算測(cè)試。

問(wèn)題："what's (3 + 5) x 12?"
結(jié)果：正確計(jì)算出答案 96
過(guò)程：先調(diào)用 add(3, 5) 得到 8，再調(diào)用 multiply(8, 12) 得到 96

實(shí)際運(yùn)行效果

通過(guò)運(yùn)行測(cè)試，可看到完整的計(jì)算流程：

問(wèn)題輸入："what's (3 + 5) x 12?"
第一次 LLM 調(diào)用：模型決定先計(jì)算加法，調(diào)用 add(3, 5)
工具執(zhí)行：MCP 服務(wù)器返回結(jié)果 8
第二次 LLM 調(diào)用：模型繼續(xù)計(jì)算乘法，調(diào)用 multiply(8, 12)
工具執(zhí)行：MCP 服務(wù)器返回結(jié)果 96
第三次 LLM 調(diào)用：模型總結(jié)最終答案

=== 獲取MCP工具 ===
可用工具: ['add', 'multiply']
  - add: Add two numbers
  - multiply: Multiply two numbers

=== 綁定工具到模型 ===
已將 2 個(gè)工具綁定到模型

=== 構(gòu)建LangGraph工作流 ===
已添加節(jié)點(diǎn): call_model (模型調(diào)用) 和 tools (工具執(zhí)行)

=== 開(kāi)始測(cè)試數(shù)學(xué)計(jì)算 ===
測(cè)試問(wèn)題: what's (3 + 5) x 12?

=== 調(diào)用LLM模型 ===
輸入消息數(shù)量: 1
最新消息: what's (3 + 5) x 12?
模型響應(yīng)類(lèi)型: AIMessage
響應(yīng)內(nèi)容: I'll help you calculate (3 + 5) × 12. Let me break this down step by step.
工具調(diào)用: 1 個(gè)
  工具 1: add - 參數(shù): {'a': 3, 'b': 5}

=== 調(diào)用LLM模型 ===
輸入消息數(shù)量: 3
最新消息: 8
模型響應(yīng)類(lèi)型: AIMessage
響應(yīng)內(nèi)容: Now I'll multiply the result (8) by 12:
工具調(diào)用: 1 個(gè)
  工具 1: multiply - 參數(shù): {'a': 8, 'b': 12}

=== 調(diào)用LLM模型 ===
輸入消息數(shù)量: 5
最新消息: 96
模型響應(yīng)類(lèi)型: AIMessage
響應(yīng)內(nèi)容: The result of (3 + 5) × 12 is **96**. 

Here's the calculation:
- First, 3 + 5 = 8
- Then, 8 × 12 = 96

=== 最終結(jié)果 ===
消息鏈長(zhǎng)度: 6
消息 1 (HumanMessage): what's (3 + 5) x 12?
消息 2 (AIMessage): I'll help you calculate (3 + 5) × 12. Let me break this down step by step.
消息 3 (ToolMessage): 8
消息 4 (AIMessage): Now I'll multiply the result (8) by 12:
消息 5 (ToolMessage): 96
消息 6 (AIMessage): The result of (3 + 5) × 12 is **96**. 

Here's the calculation:
- First, 3 + 5 = 8
- Then, 8 × 12 = 96

服務(wù)端封裝

方案概覽

本節(jié)演示如何將已有的 LangGraph 工作流封裝為一個(gè) FastMCP 服務(wù)器，對(duì)外以 MCP 標(biāo)準(zhǔn)協(xié)議提供能力。核心思路是：在 FastMCP 中注冊(cè)一個(gè)工具（process_text_with_langgraph），其內(nèi)部調(diào)用 LangGraph 的工作流，實(shí)現(xiàn)“預(yù)處理 → AI 分析 → 結(jié)果匯總”的端到端處理。

#!/usr/bin/env python3
"""
簡(jiǎn)化的 FastMCP + LangGraph 演示
展示如何在FastMCP中集成LangGraph工作流
"""

import asyncio
import os
from typing import TypedDict, List
from datetime import datetime

from fastmcp import FastMCP, Context
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage
from langchain_deepseek import ChatDeepSeek
from langchain_core.prompts import ChatPromptTemplate

# 加載環(huán)境變量
from dotenv import load_dotenv
load_dotenv()

# 定義簡(jiǎn)單的狀態(tài)類(lèi)型
class TextProcessState(TypedDict):
    input_text: str
    processed_text: str
    ai_response: str
    steps: List[str]

# 初始化DeepSeek模型
model = ChatDeepSeek(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.7
)

# 創(chuàng)建FastMCP實(shí)例
mcp = FastMCP("Simple-FastMCP-LangGraph")

def create_text_processing_graph():
    """創(chuàng)建文本處理的LangGraph工作流"""
    
    asyncdef preprocess_text(state: TextProcessState) -> TextProcessState:
        """預(yù)處理文本"""
        input_text = state["input_text"]
        
        # 簡(jiǎn)單的預(yù)處理：去除多余空格，添加時(shí)間戳
        processed = input_text.strip()
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        
        return {
            **state,
            "processed_text": processed,
            "steps": state["steps"] + [f"文本預(yù)處理完成 ({timestamp})"]
        }
    
    asyncdef generate_ai_response(state: TextProcessState) -> TextProcessState:
        """生成AI響應(yīng)"""
        processed_text = state["processed_text"]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一個(gè)專(zhuān)業(yè)的文本分析助手。請(qǐng)對(duì)用戶提供的文本進(jìn)行分析和總結(jié)，提供有價(jià)值的見(jiàn)解。"),
            ("human", "請(qǐng)分析以下文本：\n\n{text}\n\n請(qǐng)?zhí)峁?) 主要內(nèi)容總結(jié) 2) 關(guān)鍵信息提取 3) 簡(jiǎn)短評(píng)價(jià)")
        ])
        
        try:
            response = await model.ainvoke(prompt.format_messages(text=processed_text))
            ai_content = response.content
        except Exception as e:
            ai_content = f"AI處理出錯(cuò): {str(e)}"
        
        return {
            **state,
            "ai_response": ai_content,
            "steps": state["steps"] + ["AI分析完成"]
        }
    
    # 構(gòu)建工作流圖
    workflow = StateGraph(TextProcessState)
    
    # 添加節(jié)點(diǎn)
    workflow.add_node("preprocess", preprocess_text)
    workflow.add_node("ai_analyze", generate_ai_response)
    
    # 添加邊
    workflow.add_edge(START, "preprocess")
    workflow.add_edge("preprocess", "ai_analyze")
    workflow.add_edge("ai_analyze", END)
    
    # 編譯圖
    memory = MemorySaver()
    return workflow.compile(checkpointer=memory)

# 創(chuàng)建全局的LangGraph實(shí)例
text_processor = create_text_processing_graph()

@mcp.tool()
asyncdef process_text_with_langgraph(text: str, ctx: Context = None) -> str:
    """
    使用LangGraph處理文本
    
    Args:
        text: 要處理的文本內(nèi)容
    
    Returns:
        處理結(jié)果
    """
    returnawait _analyze_text(text, ctx)

# 原始工具函數(shù)（不使用裝飾器）
asyncdef _analyze_text(text: str, ctx = None) -> str:
    """內(nèi)部文本分析函數(shù)"""
    if ctx:
        await ctx.info(f"開(kāi)始分析文本: {text[:30]}...")
    
    try:
        # 初始狀態(tài)
        initial_state = {
            "input_text": text,
            "processed_text": "",
            "ai_response": "",
            "steps": []
        }
        
        # 配置
        config = {
            "configurable": {
                "thread_id": f"analyze_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
            }
        }
        
        if ctx:
            await ctx.info("執(zhí)行LangGraph工作流...")
        
        # 運(yùn)行工作流
        final_state = await text_processor.ainvoke(initial_state, config)
        
        if ctx:
            await ctx.info("分析完成")
        
        # 格式化結(jié)果
        result = f"""?? 文本分析結(jié)果

?? 原始文本:
{final_state['input_text']}

?? AI分析:
{final_state['ai_response']}

?? 處理步驟:
{' → '.join(final_state['steps'])}

? 完成時(shí)間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""
        
        return result
        
    except Exception as e:
        error_msg = f"文本分析失敗: {str(e)}"
        if ctx:
            await ctx.error(error_msg)
        return error_msg

if __name__ == "__main__":
    import sys
    
    # 服務(wù)器模式
    print("?? 啟動(dòng) Simple FastMCP + LangGraph 服務(wù)器")
    print("?? 可用工具:")
    print("   - process_text_with_langgraph: 使用LangGraph處理文本")
    print("?? 服務(wù)器地址: http://127.0.0.1:8004/mcp")
    print("?? 測(cè)試命令: python simple_fastmcp_demo.py --test")
    print("=" * 60)
    
    mcp.run(
        transport="http",
        host="127.0.0.1",
        port=8004,
        log_level="info"
    )

啟動(dòng)與驗(yàn)證
在啟動(dòng)前，請(qǐng)確保已在項(xiàng)目根目錄配置 .env 并設(shè)置 DEEPSEEK_API_KEY。

運(yùn)行代碼
```Shell
uv run langgraph_use_mcp_as_server.py

啟動(dòng)成功后，可使用 MCP Inspector 連接 http://127.0.0.1:8004/mcp，界面中將自動(dòng)發(fā)現(xiàn)并展示工具 process_text_with_langgraph。選擇該工具，輸入任意文本（例如“請(qǐng)分析這段關(guān)于 LangGraph 與 FastMCP 集成的描述，給出要點(diǎn)與建議”）進(jìn)行調(diào)用。工具將返回結(jié)構(gòu)化的文本結(jié)果，包括：

原始文本：回顯你輸入的內(nèi)容；
AI 分析：模型基于提示模版給出的總結(jié)、關(guān)鍵信息與評(píng)價(jià)；
處理步驟：包含“文本預(yù)處理完成（時(shí)間戳）→ AI 分析完成”的流水；
完成時(shí)間：本次處理的結(jié)束時(shí)間戳。

通過(guò) Inspector 調(diào)用新增的 MCP 服務(wù)器，注意需要調(diào)整超時(shí)時(shí)間

LangGraph × FastMCP 2.0 強(qiáng)強(qiáng)聯(lián)合：輕松構(gòu)建企業(yè)級(jí)AI工作流，效率翻倍！-AI.x社區(qū)

運(yùn)行結(jié)果說(shuō)明

服務(wù)器地址：http://127.0.0.1:8004/mcp（HTTP 傳輸，F(xiàn)astMCP 會(huì)輸出啟動(dòng)日志）。
可用工具：process_text_with_langgraph（用于對(duì)文本進(jìn)行預(yù)處理與 AI 分析）。
成功調(diào)用后：在 Inspector 中可看到該工具的響應(yīng)正文，包含“原始文本 / AI 分析 / 處理步驟 / 完成時(shí)間”等字段；若輸入較長(zhǎng)文本，處理步驟會(huì)顯示帶時(shí)間戳的進(jìn)度信息，便于排查與復(fù)現(xiàn)。

本文轉(zhuǎn)載自??AI 博物院?? 作者：longyunfeigu

?著作權(quán)歸作者所有，如需轉(zhuǎn)載，請(qǐng)注明出處，否則將追究法律責(zé)任

標(biāo)簽

LangGraph

FastMCP

MCP

已于2025-9-8 07:00:33修改

贊

回復(fù)