mcp = FastMCP("Math")

# Prompts
@mcp.prompt()
def example_prompt(question: str) -> str:
"""Example prompt description"""
return f"""
You are a math assistant. Answer the question.
Question: {question}
"""

@mcp.prompt()
def system_prompt() -> str:
"""System prompt description"""
return """
You are an AI assistant use the tools if needed.
"""

# Resources
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"

@mcp.resource("config://app")
def get_config() -> str:
"""Static configuration data"""
return "App configuration here"

# Tools
@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__":
mcp.run() # Run server via stdio

此代碼：

對于可流式 HTTP：

...



if __name__ == "__main__":

    mcp.run(transport="streamable-http")  # Run server via streamable-http

另存為 math_mcp_server.py

第 2 步：創建 MCP 客戶端

為了與服務器交互，我們使用 MCP 客戶端。客戶端通過 stdio 與服務器通信，允許我們列出提示、資源和工具，并調用它們。

from mcp import ClientSession, StdioServerParameters

from mcp.client.stdio import stdio_client

import asyncio



# Math Server Parameters

server_params = StdioServerParameters(

    command="python",

    args=["math_mcp_server.py"],

    env=None,

)



async def main():

    async with stdio_client(server_params) as (read, write):

        async with ClientSession(read, write) as session:

            await session.initialize()



            # List available prompts

            response = await session.list_prompts()

            print("\n/////////////////prompts//////////////////")

            for prompt in response.prompts:

                print(prompt)



            # List available resources

            response = await session.list_resources()

            print("\n/////////////////resources//////////////////")

            for resource in response.resources:

                print(resource)



            # List available resource templates

            response = await session.list_resource_templates()

            print("\n/////////////////resource_templates//////////////////")

            for resource_template in response.resourceTemplates:

                print(resource_template)



            # List available tools

            response = await session.list_tools()

            print("\n/////////////////tools//////////////////")

            for tool in response.tools:

                print(tool)



            # Get a prompt

            prompt = await session.get_prompt("example_prompt", arguments={"question": "what is 2+2"})

            print("\n/////////////////prompt//////////////////")

            print(prompt.messages[0].content.text)



            # Read a resource

            content, mime_type = await session.read_resource("greeting://Alice")

            print("\n/////////////////content//////////////////")

            print(mime_type[1][0].text)



            # Call a tool

            result = await session.call_tool("add", arguments={"a": 2, "b": 2})

            print("\n/////////////////result//////////////////")

            print(result.content[0].text)



if __name__ == "__main__":

    asyncio.run(main())

輸出

運行客戶端代碼將產生以下輸出

Processing request of type ListPromptsRequest



/////////////////prompts//////////////////

name='example_prompt' description='Example prompt description' arguments=[PromptArgument(name='question', description=None, required=True)]

name='system_prompt' description='System prompt description' arguments=[]



Processing request of type ListResourcesRequest



/////////////////resources//////////////////

uri=AnyUrl('config://app') name='get_config' description='Static configuration data' mimeType='text/plain' size=None annotations=None



Processing request of type ListResourceTemplatesRequest



/////////////////resource_templates//////////////////

uriTemplate='greeting://{name}' name='get_greeting' description='Get a personalized greeting' mimeType=None annotations=None



Processing request of type ListToolsRequest



/////////////////tools//////////////////

name='add' description='Add two numbers' inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'addArguments', 'type': 'object'} annotations=None

name='multiply' description='Multiply two numbers' inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'multiplyArguments', 'type': 'object'} annotations=None



Processing request of type GetPromptRequest



/////////////////prompt//////////////////

    You are a math assistant. Answer the question.

    Question: what is 2+2



Processing request of type ReadResourceRequest



/////////////////content//////////////////

Hello, Alice!



Processing request of type CallToolRequest



/////////////////result//////////////////

4

此輸出顯示：

對于可流式傳輸的 HTTP

from mcp.client.streamable_http import streamablehttp_client



# Math server

math_server_url = "http://localhost:8000/mcp"



async def main():

    async with streamablehttp_client(math_server_url) as (read, write, _):

        async with ClientSession(read, write) as session:

            ...

第 3 步：將 MCP 與 LangGraph 集成

LangGraph 允許我們使用基于圖形的方法構建有狀態工作流。我們可以將 MCP 客戶端與 LangGraph 集成，以創建一個使用服務器工具和提示的 AI 代理。

from typing import List

from typing_extensions import TypedDict

from typing import Annotated



from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

from langchain_google_genai import ChatGoogleGenerativeAI

from langgraph.prebuilt import tools_condition, ToolNode

from langgraph.graph import StateGraph, START, END

from langgraph.graph.message import AnyMessage, add_messages

from langgraph.checkpoint.memory import MemorySaver



from langchain_mcp_adapters.tools import load_mcp_tools

from langchain_mcp_adapters.resources import load_mcp_resources

from langchain_mcp_adapters.prompts import load_mcp_prompt

from mcp import ClientSession, StdioServerParameters

from mcp.client.stdio import stdio_client



import asyncio



# Math Server Parameters

server_params = StdioServerParameters(

    command="python",

    args=["math_mcp_server.py"],

    env=None,

)



async def create_graph(session):

    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0, api_key="your_google_api_key")



    tools = await load_mcp_tools(session)

    llm_with_tool = llm.bind_tools(tools)



    system_prompt = await load_mcp_prompt(session, "system_prompt")

    prompt_template = ChatPromptTemplate.from_messages([

        ("system", system_prompt[0].content),

        MessagesPlaceholder("messages")

    ])

    chat_llm = prompt_template | llm_with_tool



    # State Management

    class State(TypedDict):

        messages: Annotated[List[AnyMessage], add_messages]



    # Nodes

    def chat_node(state: State) -> State:

        state["messages"] = chat_llm.invoke({"messages": state["messages"]})

        return state



    # Building the graph

    graph_builder = StateGraph(State)

    graph_builder.add_node("chat_node", chat_node)

    graph_builder.add_node("tool_node", ToolNode(tools=tools))

    graph_builder.add_edge(START, "chat_node")

    graph_builder.add_conditional_edges("chat_node", tools_condition, {"tools": "tool_node", "__end__": END})

    graph_builder.add_edge("tool_node", "chat_node")

    graph = graph_builder.compile(checkpointer=MemorySaver())

    return graph



async def main():

    config = {"configurable": {"thread_id": 1234}}

    async with stdio_client(server_params) as (read, write):

        async with ClientSession(read, write) as session:

            await session.initialize()



            # Check available tools

            tools = await load_mcp_tools(session)

            print("Available tools:", [tool.name for tool in tools])



            # Check available prompts

            prompts = await load_mcp_prompt(session, "example_prompt", arguments={"question": "what is 2+2"})

            print("Available prompts:", [prompt.content for prompt in prompts])

            prompts = await load_mcp_prompt(session, "system_prompt")

            print("Available prompts:", [prompt.content for prompt in prompts])



            # Check available resources

            resources = await load_mcp_resources(session, uris=["greeting://Alice", "config://app"])

            print("Available resources:", [resource.data for resource in resources])



            # Use the MCP Server in the graph

            agent = await create_graph(session)

            while True:

                message = input("User: ")

                response = await agent.ainvoke({"messages": message}, config=config)

                print("AI: "+response["messages"][-1].content)



if __name__ == "__main__":

    asyncio.run(main())

輸出

Processing request of type ListToolsRequest

Available tools: ['add', 'multiply']



Processing request of type GetPromptRequest

Available prompts: ['\n    You are a math assistant. Answer the question.\n    Question: what is 2+2\n    ']



Processing request of type GetPromptRequest

Available prompts: ['\n    You are an AI assistant use the tools if needed.\n    ']



Processing request of type ReadResourceRequest

Processing request of type ReadResourceRequest

Available resources: ['Hello, Alice!', 'App configuration here']



Processing request of type ListToolsRequest

Processing request of type GetPromptRequest



User: Hi

AI: Hi there! How can I help you today?

User: what is 2 + 4

Processing request of type CallToolRequest

AI: 2 + 4 = 6

此輸出顯示：

第 4 步：將多個 MCP 服務器與 LangGraph 集成

我們可以使用 MultiServerMCPClient 連接到多個服務器。

創建另一個 MCP 服務器

from mcp.server.fastmcp import FastMCP



mcp = FastMCP("BMI")



# Tools

@mcp.tool()

def calculate_bmi(weight: int, height: int) -> str:

    """Calculate BMI"""

    return "BMI: "+str(weight/(height*height))



if __name__ == "__main__":

    mcp.run(transport="streamable-http")

將其另存為 bmi_mcp_server.py。該服務器：

多 MCPServer LangGraph 代碼（會話已關閉）

在下面的代碼中，我們使用 client.get_tools（） 和 client.get_prompt（）， 其中每個工具調用都有一個新的 MCP ClientSession

from typing import List

from typing_extensions import TypedDict

from typing import Annotated

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

from langchain_google_genai import ChatGoogleGenerativeAI

from langgraph.prebuilt import tools_condition, ToolNode

from langgraph.graph import StateGraph, START, END

from langgraph.graph.message import AnyMessage, add_messages

from langgraph.checkpoint.memory import MemorySaver

from langchain_mcp_adapters.client import MultiServerMCPClient

import asyncio



client = MultiServerMCPClient(

    {

        "math": {

            "command": "python",

            "args": ["math_mcp_server.py"],

            "transport": "stdio",

        },

        "bmi": {

            "url": "http://localhost:8000/mcp",

            "transport": "streamable_http",

        }

    }

)



async def create_graph():

    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0, api_key="your_google_api_key")

    tools = await client.get_tools()

    llm_with_tool = llm.bind_tools(tools)

    system_prompt = await client.get_prompt(server_name="math", prompt_name="system_prompt")

    prompt_template = ChatPromptTemplate.from_messages([

        ("system", system_prompt[0].content),

        MessagesPlaceholder("messages")

    ])

    chat_llm = prompt_template | llm_with_tool



    # State Management

    class State(TypedDict):

        messages: Annotated[List[AnyMessage], add_messages]



    # Nodes

    def chat_node(state: State) -> State:

        state["messages"] = chat_llm.invoke({"messages": state["messages"]})

        return state



    # Building the graph

    graph_builder = StateGraph(State)

    graph_builder.add_node("chat_node", chat_node)

    graph_builder.add_node("tool_node", ToolNode(tools=tools))

    graph_builder.add_edge(START, "chat_node")

    graph_builder.add_conditional_edges("chat_node", tools_condition, {"tools": "tool_node", "__end__": END})

    graph_builder.add_edge("tool_node", "chat_node")

    graph = graph_builder.compile(checkpointer=MemorySaver())

    return graph



async def main():

    config = {"configurable": {"thread_id": 1234}}

    agent = await create_graph()

    while True:

        message = input("User: ")

        response = await agent.ainvoke({"messages": message}, config=config)

        print("AI: "+response["messages"][-1].content)



if __name__ == "__main__":

    asyncio.run(main())

多 MCPServer LangGraph 代碼（持久會話）

我們可以使用 client.session 保持兩臺服務器的會話打開。

from typing import List

from typing_extensions import TypedDict

from typing import Annotated

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

from langchain_google_genai import ChatGoogleGenerativeAI

from langgraph.prebuilt import tools_condition, ToolNode

from langgraph.graph import StateGraph, START, END

from langgraph.graph.message import AnyMessage, add_messages

from langgraph.checkpoint.memory import MemorySaver

from langchain_mcp_adapters.client import MultiServerMCPClient

from langchain_mcp_adapters.tools import load_mcp_tools

from langchain_mcp_adapters.prompts import load_mcp_prompt

import asyncio



client = MultiServerMCPClient(

    {

        "math": {

            "command": "python",

            "args": ["math_mcp_server.py"],

            "transport": "stdio",

        },

        "bmi": {

            "url": "http://localhost:8000/mcp",

            "transport": "streamable_http",

        }

    }

)



async def create_graph(math_session, bmi_session):

    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0, api_key="your_google_api_key")



    math_tools = await load_mcp_tools(math_session)

    bmi_tools = await load_mcp_tools(bmi_session)

    tools = math_tools + bmi_tools

    llm_with_tool = llm.bind_tools(tools)



    system_prompt = await load_mcp_prompt(math_session, "system_prompt")

    prompt_template = ChatPromptTemplate.from_messages([

        ("system", system_prompt[0].content),

        MessagesPlaceholder("messages")

    ])

    chat_llm = prompt_template | llm_with_tool



    # State Management

    class State(TypedDict):

        messages: Annotated[List[AnyMessage], add_messages]



    # Nodes

    def chat_node(state: State) -> State:

        state["messages"] = chat_llm.invoke({"messages": state["messages"]})

        return state



    # Building the graph

    graph_builder = StateGraph(State)

    graph_builder.add_node("chat_node", chat_node)

    graph_builder.add_node("tool_node", ToolNode(tools=tools))

    graph_builder.add_edge(START, "chat_node")

    graph_builder.add_conditional_edges("chat_node", tools_condition, {"tools": "tool_node", "__end__": END})

    graph_builder.add_edge("tool_node", "chat_node")

    graph = graph_builder.compile(checkpointer=MemorySaver())

    return graph



async def main():

    config = {"configurable": {"thread_id": 1234}}

    async with client.session("math") as math_session, client.session("bmi") as bmi_session:

        agent = await create_graph(math_session, bmi_session)

        while True:

            message = input("User: ")

            response = await agent.ainvoke({"messages": message}, config=config)

            print("AI: "+response["messages"][-1].content)



if __name__ == "__main__":

    asyncio.run(main())

輸出

User: Hi

AI: Hi there! How can I help you today?

User: how many tools do you have

AI: I have 3 tools available: add, multiply, and calculate_bmi.
User: find 5 * 4
Processing request of type CallToolRequest
AI: The answer is 20.

此輸出顯示：

結論

通過將 MCP 與 LangGraph 相結合，您可以構建靈活的模塊化 AI 系統，在有狀態工作流程中利用結構化提示和工具。MCP 服務器提供了一個干凈的界面來定義 AI 功能，而 LangGraph 則協調信息流。

文章轉載自：Creating an MCP Server and Integrating with LangGraph

上一篇：

構建基于Claude MCP的天氣查詢智能體 | 實戰落地示例

下一篇：

MCP 企業級落地案例架構設計和落地實戰
#你可能也喜歡這些API文章!

我們有何不同？

API服務商零注冊

多API并行試用

數據驅動選型，提升決策效率

查看全部API→
??

熱門場景實測，選對API

#AI文本生成大模型API

對比大模型API的內容創意新穎性、情感共鳴力、商業轉化潛力

25個渠道
一鍵對比試用API 限時免費

#AI深度推理大模型API

對比大模型API的邏輯推理準確性、分析深度、可視化建議合理性

10個渠道
一鍵對比試用API 限時免費