Hummingbot 第15章：MCP 与 AI Skills

Hummingbot 的 MCP（Model Context Protocol）Server 和 AI Skills 框架为交易机器人带来了人工智能驱动的自动化和智能化能力。本章详细介绍如何将 AI 助手集成到 Hummingbot 工作流中。

Hummingbot MCP Server 介绍

什么是 MCP

MCP（Model Context Protocol）是一种开放协议，允许 AI 模型与外部工具和数据源进行交互。Hummingbot 的 MCP Server 实现了此协议，使 AI 助手能够直接控制交易机器人。

架构概览

┌──────────────┐       ┌─────────────────┐       ┌──────────────┐
│  AI 助手      │ ◄───► │ MCP Server      │ ◄───► │ Hummingbot   │
│ (Claude/等)   │       │ (hummingbot-mcp)│       │ API/实例     │
└──────────────┘       └─────────────────┘       └──────────────┘

安装 MCP Server

# 通过 pip 安装
pip install hummingbot-mcp

# 验证安装
hummingbot-mcp --version

启动 MCP Server

# 默认配置启动
hummingbot-mcp

# 指定 Hummingbot API 地址
hummingbot-mcp --api-host 127.0.0.1 --api-port 15871

# 指定 API 凭证
hummingbot-mcp --api-key your-key --api-secret your-secret

AI 助手集成

与 Claude Code 集成

将 Hummingbot MCP Server 添加到 Claude Code 的配置中：

{
  "mcpServers": {
    "hummingbot": {
      "command": "hummingbot-mcp",
      "args": [
        "--api-key", "your-api-key",
        "--api-secret", "your-api-secret"
      ]
    }
  }
}

可用工具列表

MCP Server 暴露以下工具给 AI 助手：

与 AI 对话示例

通过与配置好的 AI 助手对话，可以自然语言控制交易机器人：

用户: "当前比特币价格是多少？"

AI 助手: "我正在查询 BTC/USDT 的实时价格..."
[调用 get_ticker 工具]
"当前 BTC/USDT 价格为 $50,123.45，24小时交易量为 12,345 BTC。"

用户: "请帮我把做市策略切换到 ETH/USDT"

AI 助手: "好的，让我先查看当前配置，然后导入新配置..."
[调用 get_config 和 import_config 工具]
"策略已切换到 ETH/USDT 纯做市策略。当前价差设置为 0.5%。"

Skills 框架

Skills 概述

Hummingbot Skills 是一个可扩展的框架，允许开发者创建自定义的 AI 驱动交易能力。Skills 可以执行复杂的多步骤操作。

内置 Skills

启用 Skills

# conf_client.yml
skills:
  enabled:
    - market_making
    - risk_management
  auto_execute: false  # 手动确认后再执行

自定义 Skills 开发

Skill 基本结构

from hummingbot.skills import Skill, SkillConfig
from hummingbot.client import HummingbotClient

class MyCustomSkill(Skill):
    """自定义交易技能"""

    def __init__(self, client: HummingbotClient):
        super().__init__(client)
        self.name = "my_custom_skill"
        self.description = "自定义交易策略管理"

    async def execute(self, params: dict) -> dict:
        """
        执行技能逻辑

        参数:
            params: 包含执行参数的字典

        返回:
            包含执行结果的字典
        """
        action = params.get("action")

        if action == "analyze":
            return await self._analyze_market(params)
        elif action == "execute_trade":
            return await self._execute_trade(params)
        else:
            return {"status": "error", "message": f"未知操作: {action}"}

    async def _analyze_market(self, params: dict) -> dict:
        """分析市场状况"""
        pair = params.get("pair", "BTC-USDT")
        exchange = params.get("exchange", "binance")

        orderbook = await self.client.get_orderbook(pair, exchange)
        ticker = await self.client.get_ticker(pair, exchange)

        spread = orderbook["asks"][0][0] - orderbook["bids"][0][0]
        spread_pct = spread / ticker["last_price"] * 100

        return {
            "status": "ok",
            "pair": pair,
            "spread_pct": round(spread_pct, 3),
            "bid_depth": sum(b[1] for b in orderbook["bids"][:5]),
            "ask_depth": sum(a[1] for a in orderbook["asks"][:5]),
            "recommendation": "做市" if spread_pct > 0.1 else "观望"
        }

    async def _execute_trade(self, params: dict) -> dict:
        """执行交易操作"""
        # 实际交易逻辑
        return {"status": "ok", "message": "交易执行完成"}

注册自定义 Skill

# skills_registry.py
from hummingbot.skills import register_skill
from my_skills.my_custom_skill import MyCustomSkill

register_skill("my_custom_skill", MyCustomSkill)

Skill 配置

{
    "skills": {
        "my_custom_skill": {
            "enabled": true,
            "config": {
                "max_order_size": 0.1,
                "min_spread": 0.001,
                "auto_rebalance": false
            }
        }
    }
}

与 LangChain 集成

使用 LangChain Agent

from langchain.agents import Tool, AgentExecutor
from langchain.agents import create_react_agent
from langchain_anthropic import ChatAnthropic
from hummingbot_mcp import HummingbotMCPClient

# 初始化 MCP 客户端
mcp_client = HummingbotMCPClient(
    api_key="your-key",
    api_secret="your-secret"
)

# 定义工具
tools = [
    Tool(
        name="get_market_status",
        func=lambda pair: mcp_client.get_ticker(pair, "binance"),
        description="获取交易对的市场行情，输入格式: 'BTC-USDT'"
    ),
    Tool(
        name="check_balances",
        func=lambda _: mcp_client.get_balances(),
        description="查询账户余额"
    ),
    Tool(
        name="start_trading",
        func=lambda config: mcp_client.import_config(config),
        description="导入并启动交易策略配置文件"
    )
]

# 创建 LangChain Agent
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

# 执行自然语言指令
result = agent_executor.invoke({
    "input": "检查 BTC/USDT 市场状况和我的账户余额，如果条件合适就开始做市"
})
print(result["output"])

与 Claude 等 AI 工具联动

自动化监控示例

import asyncio
from hummingbot_mcp import HummingbotMCPClient

async def ai_monitor():
    client = HummingbotMCPClient(api_key="key", api_secret="secret")

    while True:
        # 定期获取状态
        status = await client.get_status()
        balances = await client.get_balances()

        # 检查异常情况
        alerts = []
        for asset, info in balances["balances"].items():
            if float(info["total"]) < 0.001:
                alerts.append(f"{asset} 余额不足: {info['total']}")

        if alerts:
            print("告警:", "; ".join(alerts))

        await asyncio.sleep(60)

asyncio.run(ai_monitor())

最佳实践

1. 权限控制：为 MCP Server 使用专用的 API Key，仅授予必要权限
2. 审核日志：所有 AI 发起的操作都会被记录，定期审计
3. 人工确认：大额交易操作建议设置 auto_execute: false
4. 错误处理：AI 工具调用可能失败，确保有回退逻辑