langchain结合mcp服务 管理mcp服务增删改查

快速方便的给langchain的agent添加mcp服务的能力

pip install mcpstore

这个库做了很多的生命周期的管理和处理,还有很多额外的功能。思路就是初始化一个store对象,然后库库往这个store上面注册mcp服务。

store = MCPStore.setup_store()

store.for_store().add_service({"name": "mcpstore-wiki", "url": "https://mcpstore.wiki/mcp"})

store.for_store().wait_service("mcpstore-wiki")

需要用他的tools的时候直接for_langchain().list_tools()就是langchain可以直接使用的格式了

tools = store.for_store().for_langchain().list_tools()

几行代码就可以直接使用注册的mcp服务了,这个tools列表是完全兼容langchain的,可以和我们已经通过@tool定义好的方法直接使用列表加法添加.

作者还提供了一个可以直接运行的例子:

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from mcpstore import MCPStore
store = MCPStore.setup_store()
store.for_store().add_service({"name":"mcpstore-wiki","url":"https://mcpstore.wiki/mcp"})
store.for_store().wait_service("mcpstore-wiki")
tools = store.for_store().for_langchain().list_tools()

llm = ChatOpenAI(
    temperature=0, model="deepseek-chat",
    openai_api_key="sk-****",
    openai_api_base="httpss://api.deepseek.com"
)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手,回答的时候带上表情"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
query = "北京的天气怎么样?"
print(f"\n   🤔: {query}")
response = agent_executor.invoke({"input": query})
print(f"   🤖 : {response['output']}")

只需要替换模型就可以直接使用了,本人亲测还是好用的。

添加本地服务的时候:

demo_mcp = {
  "mcpServers": {
    "howtocook": {
      "command": "npx",
      "args": [
        "-y",
        "howtocook-mcp"
      ]
    }
  }
}
store = MCPStore.setup_store()

store.for_store().add_service(demo_mcp)

store.for_store().wait_service("howtocook")

wait_service可以设置一个超时的时间,对于需要长时间启动的本地服务可以设置长一点,或者使用他的异步方法_sync后缀的方法。

可视化管理界面

作者的GitHub仓库上有一个开源的demo Vue项目,可以通过简单的几行代码开启API模式,搭配Vue可以有一个前端点点点的维护mcp服务管理平台:

image-20250721212359929

通过一段简单的代码快速启动后端:

from mcpstore import MCPStore
prod_store = MCPStore.setup_store()
prod_store.start_api_server(
    host='0.0.0.0',
    port=18200
)

通过 https://mcpstore.wiki/web_demo/dashboard 体验在线示例

通过 https://doc.mcpstore.wiki/ 可以查看详细的使用文档

🔧 注册服务

对于服务的增删改查,作者尽可能的兼容各种各样的格式,保证用户可以不需要太过于专注格式

只要不冲突并且关键的配置对就行,所以支持市面上常见的字典和列表等等格式

主流的Client的json配置文件都可以直接使用,非常方便。

🌐 远程服务注册

通过 URL 添加远程 MCP 服务:

store.for_store().add_service({
    "name": "mcpstore-wiki",
    "url": "https://mcpstore.wiki/mcp",
    "transport": "streamable-https"
})
  • transport: 可选字段,可以自动推断传输协议 (streamable-https, sse)

等待服务的状态转变

mcpstore里,有一个wait_service()函数,等待一个服务转变到指定的状态,可以设置超时时间

store.for_store().add_service(demo_mcp)

store.for_store().wait_service("howtocook")

mcpstore内置了几个生命周期来管理服务,用户只需要注册,mcpstore会异步的更新维护服务

从在线市场添加服务

mcpstore可以通过from_market参数通过服务名,直接添加已经发布过的在线的mcp服务,用户可以指定增量更新市场数据

service_name = "quickchart"
market_env = {}

store.for_store().add_service(from_market=service_name, market_env=market_env)

store.for_store().wait_service(service_name,timeout=20,status="healthy")

📄 支持的注册格式

使用标准 MCP 配置格式(主流IDE支持的mcp.json的格式):

store.for_store().add_service({
  "mcpServers": {
    "mcpstore-wiki": {
      "url": "https://mcpstore.wiki/mcp"
    },
    "howtocook": {
      "command": "npx",
      "args": [
        "-y",
        "howtocook-mcp"
      ]
    }
  }
})

💻 支持各种各样的本地服务进程

启动本地 MCP 服务进程:

# Python 服务
store.for_store().add_service({
    "name": "local_assistant",
    "command": "python",
    "args": ["./assistant_server.py"],
    "env": {"DEBUG": "true", "API_KEY": "your_key"},
    "working_dir": "/path/to/service"
})

# Node.js 服务
store.for_store().add_service({
    "name": "node_service",
    "command": "node",
    "args": ["server.js", "--port", "8080"],
    "env": {"NODE_ENV": "production"}
})

# 可执行文件
store.for_store().add_service({
    "name": "binary_service",
    "command": "./mcp_server",
    "args": ["--config", "config.json"]
})

🎯 总结

该项目更新比较频繁,感觉作者也在全力做一个直观好用的MCP管理,对于Langchain集成MCP服务是一个不错的选择。

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐