目录

前言

langchain调用MCP

动态生成MCPServer的tools

需求背景

初步实现思路

上代码

测试验证

结语

下次安排

---

前言

上次MCP文章说了后面会再写写如何用langchain更加高效的实现mcp调用。还有一个是，如何将现有应用api，如何转换为mcp提供调用。

这次我们就安排上。

langchain调用MCP

对于MCP Server来说没啥区别，还是使用mcp类库构建Server或直接使用第三方Server。langchain主要是可以简化调用MCP Server的过程。

这里会用到langchain的langchain_mcp_adapters类库，需要pip install安装下

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
# from langchain_community.chat_models.openai import ChatOpenAI
from langchain_openai import ChatOpenAI

import os
from dotenv import load_dotenv

load_dotenv()


async def main():

    client = MultiServerMCPClient(
        {
            "amap-amap-sse": {
                "url": "https://mcp.amap.com/sse?key=yourkey",
                "transport": "sse",
            }
        }
    )
    
    tools = await client.get_tools()
    base_url = os.getenv("QWEN_BASE_URL")
    api_key = os.getenv("QWEN_API_KEY")
    llm = ChatOpenAI(base_url=base_url, api_key=api_key, model="qwen-plus-2025-04-28")
    agent = create_react_agent(llm, tools)
    # math_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
    weather_response = await agent.ainvoke({"messages": "what is the weather in 北京?"},stream_mode="chunks")
    
    # res = await llm.ainvoke("what's (3 + 5) x 12?")
    # print(res)
    
    # print(math_response)
    print("==========")
    print(weather_response)
    
    # async for chunk in weather_response:
    #     print(chunk)


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

我们简单分析下这个代码，通过一个MCP配置即可创建client对象，然后获取tools。接下来就是创建一个agent，传入tools就可以调用MCP了。

它的好处自然是代码很简单，但是做了封装自由度就小了。而且我试了下，用了langchain的方式，没法实现一步步工具调用、最终结果的流式输出。这种需要和前端交互的情况，显然是不合适的。

果断放弃。

动态生成MCPServer的tools

需求背景

业务系统中随处可见的api接口，可能还是分散在不同的系统、服务下。当有了大模型能力，想要集成这些api的时候，其实也是个麻烦事。

传统的做法，都需要在AI应用中，再写一个本地方法去调用相关api，然后就可以和function call功能对接上了；而且各种api调用方式不尽相同；管理上也很凌乱。

然而有了MCP，我可以搞个MCP Server，直接相当于做了业务api的整合。通过MCP方式和大模型交互的时候，也避免了传统方式的高耦合问题。

初步实现思路

业务api数量巨大，一个个都写一个对应的方法作为tool，很不现实，而且多数人都不想这么干吧。于是就想能不能动态生成呢？

我们如果写过MCPServer就知道定义了一个方法就可以通过@mcp.tool()标记，自动将方法添加到MCP tools中。当然MCP也提供了手动添加tool的方法，mcp.add_tool()

基于这个，我们可以先把相关api做一个配置，然后遍历读取配置，生成一个动态方法，再手动add_tool()，我们的愿望基本就可以达成了。

上代码

这个是我初步尝试的两个api配置，只要可以通了，一切就简单了

functions_config = [
    {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "params": {"city": str},
        "return_type": dict,
        "api_url": "http://localhost:15002/get_weather",
    },
    {
        "name": "get_bmi",
        "description": "计算bmi",
        "params": {"height_m": int, "weight_kg": int},
        "return_type": dict,
        "api_url": "http://localhost:15002/get_bmi",
    },
]

接下来实现一个动态生成方法的方法，并遍历配置添加tool

import requests

from inspect import Parameter, Signature

def create_dynamic_function(api_url, params):
    # 动态生成函数签名
    parameters = [
        Parameter(name=param_name, kind=Parameter.POSITIONAL_OR_KEYWORD, default=None)
        for param_name in params.keys()
    ]
    signature = Signature(parameters)

    def dynamic_function(*args, **kwargs):
        try:
            # 绑定参数到签名
            bound_args = signature.bind(*args, **kwargs)
            bound_args.apply_defaults()

            # 提取绑定后的参数
            kwargs = bound_args.arguments

            print(f"Calling API: {api_url} with params: {kwargs}")
            response = requests.post(api_url, json=kwargs)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"API call failed: {e}")
            return {"error": str(e)}

    # 设置函数签名
    dynamic_function.__signature__ = signature
    return dynamic_function


def register_dynamic_tools(mcp, functions_config):
    for config in functions_config:
        name = config["name"]
        description = config.get("description", "")
        params = config.get("params", {})
        return_type = config.get("return_type", None)
        api_url = config["api_url"]

        print(f"Registering tool: {name}, Params: {params}")

        # 创建动态函数
        dynamic_function = create_dynamic_function(api_url, params)

        # 设置函数名和文档
        dynamic_function.__name__ = name
        dynamic_function.__doc__ = description

        # 设置类型注解
        dynamic_function.__annotations__ = {**params, "return": return_type}

        # 添加到 mcp 工具中
        mcp.add_tool(dynamic_function)

    print("所有动态函数已注册完成")

其他代码就是常规的MCP Server了

from mcp.server.fastmcp import FastMCP
import datetime

mcp = FastMCP(
    name="myMCP",
    host="0.0.0.0",
    port=8890,
    description="动态mcp",
    sse_path="/sse",
)

if __name__ == "__main__":
    # 初始化并运行服务器
    try:
        print("Starting server...")
        mcp.run(transport="sse")
        
    except Exception as e:
        print(f"Error: {e}")

测试验证

我们创建一个client直接连接这个MCP Server，并调用动态生成的tool，通过输出看是可以成功调用的。

后续就是将api配置换成实际的api去验证了，包括要加上鉴权等功能。核心是一步一步验证，一步一步走。

结语

本次主要讲了如何通过langchain调用MCP，以及如何动态生成MCP tool，相对简单。

下次安排

之前版本调用tool是提示词写的（就是把tool取出来都拼到了提示词），随着添加的mcp工具越来越多，维护更加困难，而且通过提示词来控制，随着工具越来越多导致复杂性增加，整体的不可靠性也会增加。后续会改为使用function call方式，这样的情况，系统提示词都不需要咋写，大模型根据用户输入+定义的tools自己判断就好，我们不需要干预。

再一个是实现调用工具、最终结果的流式输出，这个对于需要和前端交互的体验是很好的。