项目简介

LangChain 通用工具服务器(Universal Tool Server)旨在解耦工具开发与智能体(Agent)开发。它提供一个集中的后端服务,用于注册、管理和执行各种外部工具(如数据检索、API调用等)。智能体可以通过连接到这个服务器来发现和使用工具,无需直接集成工具代码,简化了配置、认证和部署流程。

该项目最初是作为通过 HTTP 暴露 LangChain 工具的解决方案,后来集成了对 Model Context Protocol (MCP) SSE 传输协议的可选支持。虽然项目主页推荐使用其他专门的 MCP 实现(如 FastMCP),但 Universal Tool Server 确实包含一个可运行的 MCP 服务器实现(通过启用特定配置)。

主要功能点

  • 工具托管与管理: 集中注册和管理多种外部工具。
  • 工具发现与调用: 客户端可以列出可用的工具,并按名称调用工具,传递参数并接收结果。
  • LangChain 集成: 无缝集成现有的 LangChain 工具。
  • 灵活部署: 可作为无状态 Web 服务部署,易于扩展。
  • 内置认证与权限(主要针对非 MCP 接口): 提供认证机制,可以为不同工具设置访问权限。
  • 可选的 MCP SSE 支持: 通过配置可启用基于 SSE 传输的 MCP 接口,支持 MCP 客户端连接(注意:MCP 接口目前不支持认证)。

安装步骤

安装非常简单,通过 pip 命令即可完成:

pip install universal-tool-server

通常,您还需要安装客户端库来连接服务器:

pip install universal-tool-client

服务器配置 (针对 MCP 客户端)

Universal Tool Server 的 MCP 功能需要通过在服务器启动时启用。对于 MCP 客户端来说,它通常需要一个 JSON 格式的配置来了解如何启动或连接到 MCP 服务器。

以下是启动一个启用 MCP SSE 功能的 Universal Tool Server 并供 MCP 客户端连接的典型配置信息。请根据您实际的服务器文件和运行环境进行调整:

{
  "name": "Universal Tool Server MCP SSE",
  "description": "启动 Universal Tool Server 并启用 MCP SSE 端点",
  "command": ["uvicorn"],
  "args": [
    "your_server_file:app",
    "--host", "127.0.0.1",
    "--port", "8002",
    "--reload"
  ],
  "note": "请将 'your_server_file:app' 替换为你实际的服务器文件及应用对象(例如,如果服务器代码在 main.py 中,应用对象名为 app,则为 'main:app')。\n确保在创建 'app' 对象时设置 enable_mcp=True。\n服务器启动后,MCP 客户端应连接到 http://127.0.0.1:8002/mcp/sse。"
}
  • 'name': 服务器的名称,用于在客户端界面显示。
  • 'description': 对服务器功能的简要描述。
  • 'command': 用于启动服务器的可执行程序命令(例如 'uvicorn')。
  • 'args': 传递给 'command' 的参数。这些参数配置了服务器监听的地址和端口,以及加载的应用对象。请确保您的应用代码文件和应用对象名称与此处匹配。
  • 'note': 重要提示信息,解释如何准备您的服务器代码(启用 MCP)以及客户端需要连接的具体地址。

基本使用方法

  1. 编写服务器代码: 创建一个 Python 文件(例如 'server.py'),定义并添加您的工具,并启用 MCP 支持。

    # server.py
    from universal_tool_server import Server
    
    # 创建Server实例并启用MCP
    app = Server(enable_mcp=True)
    
    # 使用装饰器添加工具
    @app.add_tool()
    async def echo(msg: str) -> str:
        """Echo a message."""
        return msg + "!"
    
    @app.add_tool
    async def add(x: int, y: int) -> int:
        """Add two numbers."""
        return x + y
    
    # 如果直接运行此文件,可以使用 uvicorn 启动服务
    if __name__ == "__main__":
        import uvicorn
        uvicorn.run("__main__:app", reload=True, port=8002)
  2. 启动服务器: 在终端中运行您的服务器文件(如果使用了 'if name == "main":' 块):

    python server.py
    # 或直接使用 uvicorn
    # uvicorn server:app --reload --port 8002
  3. MCP 客户端连接: 使用支持 MCP 的客户端库(例如 'mcp' 库)连接到服务器暴露的 MCP SSE 端点(例如 'http://localhost:8002/mcp/sse')。

    # client.py (使用 mcp 库)
    import asyncio
    import sys
    from mcp import ClientSession
    from mcp.client.sse import sse_client # SSE 传输协议客户端
    
    async def main():
        # 确保替换为您的服务器实际地址和 MCP SSE 路径
        url = "http://localhost:8002/mcp/sse"
    
        print(f"Connecting to MCP SSE server at {url}")
    
        try:
            async with sse_client(url=url) as streams:
                async with ClientSession(streams[0], streams[1]) as session:
                    await session.initialize() # 初始化会话
                    tools = await session.list_tools() # 列出可用工具
                    print("Available Tools:", tools)
    
                    # 调用 'echo' 工具
                    result = await session.call_tool("echo", {"msg": "Hello from MCP client!"})
                    print("Tool call result:", result)
    
        except Exception as e:
            print(f"An error occurred: {e}")
    
    if __name__ == "__main__":
        # 需要安装 mcp 库: pip install mcp
        asyncio.run(main())
  4. 运行客户端: 在另一个终端运行客户端代码:

    python client.py

客户端将连接到服务器,列出工具,并调用您定义的 'echo' 工具。

信息

分类

AI与计算