< >
Home » OpenClaw-AI助手入门教程 » OpenClaw-AI助手入门教程-配置mcp-endpoint-server云注册中心方便接入mcp工具

OpenClaw-AI助手入门教程-配置mcp-endpoint-server云注册中心方便接入mcp工具

说明:

  • mcp-endpoint-server是什么？

在开始部署之前，我们先理解一下mcp-endpoint-server在整个架构中的位置。

mcp-endpoint-server是一个基于WebSocket的MCP注册中心，由xinnan-tech团队开发，主要作用是作为小智AI设备与MCP工具之间的桥梁。它的核心设计理念是：让小智设备端能够通过语音调用各种MCP服务。

  • 为什么需要它？

如果你接触过小智AI，你可能知道小智设备端需要通过WebSocket连接到小智官方服务器。而mcp-endpoint-server提供了一个统一的接入点，设备端可以连接到这里，然后通过它调用各种注册的MCP工具（如计算器、高德地图、Home Assistant等）。

  • 它的优势在于：

高性能：基于FastAPI和WebSocket的异步架构，支持高并发连接
连接管理：智能管理WebSocket连接，支持连接清理
监控统计：提供连接统计和健康检查接口
消息转发：自动转发工具端和小智端之间的消息
配置灵活：支持配置文件管理，易于部署和维护

  • 简单来说，你部署好mcp-endpoint-server后，小智设备端就可以连接上来，然后通过它调用你本地的MCP服务（比如我们之前配置的计算器）。而且你的MCP服务无需部署在公网环境，这极大方便了开发者在各种网络环境下的使用。

  • 工作流程：

设备端：小智设备（或运行mcp_pipe.py的机器）作为WebSocket客户端，主动连接到mcp-endpoint-server

注册工具：设备端将自己本地的MCP服务（如计算器）通过stdio管道与mcp_pipe.py连接，mcp_pipe.py负责将WebSocket消息转换为stdio通信

服务端管理：mcp-endpoint-server管理所有设备的WebSocket连接，并转发来自小智官方服务器的请求

调用链路：小智官方服务器 → mcp-endpoint-server → 设备端 → 本地MCP服务

这种设计的巧妙之处在于：设备端的MCP服务可以运行在内网，不需要公网IP，只需要设备端能主动连接到公网的mcp-endpoint-server即可

docker方式安装mcp-endpoint-server：

git clone https://github.com/xinnan-tech/mcp-endpoint-server.git
# 进入本项目源码根目录
cd mcp-endpoint-server

# 清除缓存
docker compose -f docker-compose.yml down
docker stop mcp-endpoint-server
docker rm mcp-endpoint-server
docker rmi ghcr.nju.edu.cn/xinnan-tech/mcp-endpoint-server:latest

# 启动docker容器
docker compose -f docker-compose.yml up -d
  • 第一次安装会需要一点时间
  • 启动完成之后会出现需要查看
# 查看日志
docker logs -f mcp-endpoint-server
250705 INFO-=====下面的地址分别是智控台/单模块MCP接入点地址====
250705 INFO-智控台MCP参数配置: http://172.22.0.2:8004/mcp_endpoint/health?key=abc
250705 INFO-单模块部署MCP接入点: ws://172.22.0.2:8004/mcp_endpoint/mcp/?token=def
250705 INFO-=====请根据具体部署选择使用，请勿泄露给任何人======
  • 假设你的本机地址是127.0.0.1
  • 如上，输出mcp接入点是中ws://127.0.0.1:8004/mcp_endpoint/mcp/?token=def就是你的MCP接入点地址。
  • 这个MCP接入点地址很重要，你等一下会用到
  • 简单验证
curl http://127.0.0.1:8004/health

部署一个计算器验证完整功能：

  • 部署虾哥写的计算器项目,安装
https://github.com/78/mcp-calculator
# 进入项目目录
cd mcp-calculator

conda remove -n mcp-calculator --all -y
conda create -n mcp-calculator python=3.10 -y
conda activate mcp-calculator

pip install -r requirements.txt
  • 建立执行脚本：
  • 使用自己正确的token测试
export MCP_ENDPOINT=ws://127.0.0.1:8004/mcp_endpoint/mcp/?token=abc
python mcp_pipe.py calculator.py
  • 启动后mcp-endpoint-server日志会出现类似这样的内容，

250705 -INFO-正在初始化MCP接入点: wss://2662r3426b.vicp.fun/mcp_e
250705 -INFO-发送MCP接入点初始化消息
250705 -INFO-MCP接入点连接成功
250705 -INFO-MCP接入点初始化成功
250705 -INFO-统一工具处理器初始化完成
250705 -INFO-MCP接入点服务器信息: name=Calculator, version=1.9.4
250705 -INFO-MCP接入点支持的工具数量: 1
250705 -INFO-所有MCP接入点工具已获取，客户端准备就绪
250705 -INFO-工具缓存已刷新
250705 -INFO-当前支持的函数列表: [ 'get_time', 'get_lunar', 'play_music', 'get_weather', 'handle_exit_intent', 'calculator']

建立测试脚本进行计算

  • 建立脚本：vim robot_example_cal.py
  • 脚本内容：
#!/usr/bin/env python3
"""
MCP Robot Client 示例
适配 xinnan-tech/mcp-endpoint-server + mcp-calculator
支持 JSON-RPC initialize / tools/list / tools/call
"""

import asyncio
import json
import base64
import argparse
import urllib.parse
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
import websockets

# ==================== Token 生成 ====================

def pad_key(key: str) -> bytes:
    key_bytes = key.encode("utf-8")
    if len(key_bytes) in (16, 24, 32):
        return key_bytes
    padded = bytearray(32)
    padded[:min(len(key_bytes), 32)] = key_bytes[:32]
    return bytes(padded)


def encrypt_token(key: str, agent_id: str) -> str:
    plain_text = json.dumps({"agentId": agent_id}, ensure_ascii=False)
    key_bytes = pad_key(key)
    cipher = AES.new(key_bytes, AES.MODE_ECB)
    padded_data = pad(plain_text.encode("utf-8"), AES.block_size)
    encrypted_bytes = cipher.encrypt(padded_data)
    return base64.b64encode(encrypted_bytes).decode("ascii")


# ==================== MCP Client ====================

class MCPRobotClient:
    def __init__(self, uri):
        self.uri = uri
        self.request_id = 0

    def next_id(self):
        self.request_id += 1
        return self.request_id

    async def send_request(self, websocket, method, params=None):
        req_id = self.next_id()
        request = {
            "jsonrpc": "2.0",
            "id": req_id,
            "method": method,
            "params": params or {}
        }
        await websocket.send(json.dumps(request, ensure_ascii=False))
        print(f"发送: {request}")
        return req_id

    async def receive_loop(self, websocket):
        try:
            async for message in websocket:
                data = json.loads(message)
                print(f"收到: {data}")
        except websockets.ConnectionClosed:
            print("WebSocket 连接关闭")

    async def run(self):
        async with websockets.connect(self.uri) as websocket:
            print(f"已连接 {self.uri}")

            # 启动接收任务
            receive_task = asyncio.create_task(self.receive_loop(websocket))

            # initialize
            await self.send_request(websocket, "initialize", {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {
                    "name": "robot-example",
                    "version": "1.0"
                }
            })

            await asyncio.sleep(0.5)

            # tools/list
            await self.send_request(websocket, "tools/list", {})

            await asyncio.sleep(0.5)

            # 调用 mcp-calculator
            await self.send_request(websocket, "tools/call", {
                "name": "calculator",
                "arguments": {
                    "python_expression": "10 + 5 * 2"
                }
            })

            # 等待一些响应
            await asyncio.sleep(5)
            receive_task.cancel()


# ==================== CLI ====================

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1", help="服务器地址")
    parser.add_argument("--port", type=int, default=8004, help="服务器端口")
    parser.add_argument("--agent-id", default="robot-client", help="Agent ID")
    parser.add_argument("--key", help="AES 密钥（用于生成 token）")
    parser.add_argument("--token", help="直接指定完整 token（优先级高于 --key）")

    args = parser.parse_args()

    # ====================
    # Token 解析逻辑
    # ====================
    if args.token:
        token = args.token
        print("使用直接提供的 token")
        uri = f"ws://{args.host}:{args.port}/mcp_endpoint/call/?token={token}"
    elif args.key:
        token = encrypt_token(args.key, args.agent_id)
        print(f"使用 key 生成 token: {token}")
        encoded_token = urllib.parse.quote(token)
        uri = f"ws://{args.host}:{args.port}/mcp_endpoint/call/?token={encoded_token}"
    else:
        print("必须提供 --token 或 --key")
        exit(1)

    print(f"连接地址: {uri}")

    client = MCPRobotClient(uri)
    asyncio.run(client.run())
  • 运行脚本
python robot_example_cal.py --token "def="
  • 使用自己正确的token测试，效果如下
使用直接提供的 token
连接地址: ws://127.0.0.1:8004/mcp_endpoint/call/?token=def=
已连接 ws://127.0.0.1:8004/mcp_endpoint/call/?token=def=
发送: {'jsonrpc': '2.0', 'id': 1, 'method': 'initialize', 'params': {'protocolVersion': '2024-11-05', 'capabilities': {}, 'clientInfo': {'name': 'robot-example', 'version': '1.0'}}}
收到: {'jsonrpc': '2.0', 'id': 1, 'result': {'protocolVersion': '2024-11-05', 'capabilities': {'experimental': {}, 'prompts': {'listChanged': False}, 'resources': {'subscribe': False, 'listChanged': False}, 'tools': {'listChanged': True}, 'tasks': {'list': {}, 'cancel': {}, 'requests': {'tools': {'call': {}}, 'prompts': {'get': {}}, 'resources': {'read': {}}}}}, 'serverInfo': {'name': 'Calculator', 'version': '2.14.5'}}}
发送: {'jsonrpc': '2.0', 'id': 2, 'method': 'tools/list', 'params': {}}
收到: {'jsonrpc': '2.0', 'id': 2, 'result': {'tools': [{'name': 'calculator', 'description': "For mathamatical calculation, always use this tool to calculate the result of a python expression. You can use 'math' or 'random' directly, without 'import'.", 'inputSchema': {'properties': {'python_expression': {'type': 'string'}}, 'required': ['python_expression'], 'type': 'object'}, 'outputSchema': {'additionalProperties': True, 'type': 'object'}, '_meta': {'_fastmcp': {'tags': []}}}]}}
发送: {'jsonrpc': '2.0', 'id': 3, 'method': 'tools/call', 'params': {'name': 'calculator', 'arguments': {'python_expression': '10 + 5 * 2'}}}
收到: {'jsonrpc': '2.0', 'id': 3, 'result': {'content': [{'type': 'text', 'text': '{"success":true,"result":20}'}], 'structuredContent': {'success': True, 'result': 20}, 'isError': False}}
  • 看到这样的信息，说明部署正确

纠错,疑问,交流: 请进入讨论区或 点击加入Q群

获取最新文章: 扫一扫右上角的二维码加入“创客智造”公众号


标签: openclaw-ai助手入门教程