【保姆级教程】用 小智AI 实现 语音控制 OpenClaw(附开源代码示例)

openclaw gateway status

（无视我这一堆输出，只要服务正常就可以）

2.2 由于这里可以选择不同的模式让小智发送消息给OpenClaw，这里选择Token模式，所以我们要先拿到 Token

openclaw dashboard

"gateway": {  "mode": "local",  "bind": "lan",  "http": {    "endpoints": {      "chatCompletions": {        "enabled": true      }    }  },  "auth": {    "mode": "token",    "token": "放入你的token在这里"   // 后面会用到  }}

pkill -f openclaw-gatewayopenclaw gateway restart

curl -X POST http://127.0.0.1:18789/v1/chat/completions \  -H "Content-Type: application/json" \  -H "Authorization: Bearer 你的token" \  -d '{    "model": "openclaw",    "messages": [{"role": "user", "content": "你好"}]  }'

from mcp.server.fastmcp import FastMCPimport requestsimport loggingimport osimport timeimport threadinglogging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')mcp = FastMCP("OpenClaw")@mcp.tool()def ask_openclaw(instruction: str) -> str:    """    【最重要触发规则】    用户只要提到以下任意关键词，就必须立即、无条件调用 ask_openclaw 这个工具：    龙虾、小龙虾、OpenClaw、openclaw、给龙虾、问龙虾、让龙虾、叫龙虾、找龙虾、龙虾帮我    注意：    - 飞书、QQ、微信、企业微信等只是龙虾的发送渠道之一    - 用户说“通过飞书”“发到QQ”“用微信发”等时，也要调用本工具，由龙虾自己决定如何处理    - instruction 必须是用户说的**完整原话**，一个字都不允许修改或省略    作用：把用户指令完整转发给本地 OpenClaw 智能体执行。    """    logging.info(f"收到小智云端任务 → 龙虾: {instruction}")    token = os.getenv("OPENCLAW_GATEWAY_TOKEN",                      "你的TOKEN")    try:        # 立即返回提示，防止小智判定超时        prompt_reply = "好的，已转给龙虾处理 🦞\n它正在思考，请稍等几秒，我会把它的回复告诉你。"        logging.info("已立即返回提示给小智，避免超时")        # 后台异步调用 OpenClaw（真正执行任务）        threading.Thread(            target=call_openclaw_async,            args=(instruction, token),            daemon=True        ).start()        return prompt_reply    except Exception as e:        logging.error(f"ask_openclaw 工具异常: {str(e)}")        return "转给龙虾时发生错误，请稍后再试。"def call_openclaw_async(instruction: str, token: str):    """后台真正调用 OpenClaw，让龙虾处理（包括飞书、QQ、微信等渠道）"""    tr        response = requests.post(            "http://127.0.0.1:18789/v1/chat/completions",            headers={                "Content-Type": "application/json",                "Authorization": f"Bearer {token}"            },            json={                "model": "openclaw",                "messages": [{"role": "user", "content": instruction}],                "stream": False,                "temperature": 0.7,                "max_tokens": 1500            },            timeout=90        )        if response.status_code == 200:            data = response.json()            content = data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()            if content:                logging.info(f"龙虾实际回复成功（长度 {len(content)}）: {content[:120]}...")                # TODO: 未来可以在这里把龙虾的回复再推送回小智（需要额外机制）            else:                logging.warning("龙虾返回了空回复")        else:            logging.error(f"OpenClaw 返回 HTTP 错误: {response.status_code} - {response.text[:200]}")    except requests.exceptions.Timeout:        logging.error("调用 OpenClaw 超时")    except Exception as e:        logging.error(f"后台调用 OpenClaw 失败: {str(e)}", exc_info=True)if __name__ == "__main__":    mcp.run(transport="stdio")

 MCP_ENDPOINT=wss://api.xiaozhi.me/mcp/?token=eyxx

async def pipe_process_stderr_to_terminal(process, target):    """Read data from process stderr and print to terminal"""    try:        while True:            # Read data from process stderr            data = await asyncio.to_thread(process.stderr.readline)            if not data:  # If no data, the process may have ended                logger.info(f"[{target}] Process has ended stderr output")                break            # Print stderr data to terminal (in text mode, data is already a string)            sys.stderr.write(data)            sys.stderr.flush()    except Exception as e:        logger.error(f"[{target}] Error in process stderr pipe: {e}")        raise  # Re-throw exception to trigger reconnectiondef build_server_command(target=None):    """Build [cmd,...] and env for the server process for a given target.    Priority:    - If target matches a server in config.mcpServers: use its definition    - Else: treat target as a Python script path (back-compat)    If target is None, read from sys.argv[1].    """    if target is None:        assert len(sys.argv) >= 2, "missing server name or script path"        target = sys.argv[1]    cfg = load_config()    servers = cfg.get("mcpServers", {}) if isinstance(cfg, dict) else {}    if target in servers:        entry = servers[target] or {}        if entry.get("disabled"):            raise RuntimeError(f"Server '{target}' is disabled in config")        typ = (entry.get("type") or entry.get("transportType") or "stdio").lower()        # environment for child process        child_env = os.environ.copy()        for k, v in (entry.get("env") or {}).items():            child_env[str(k)] = str(v)        if typ == "stdio":            command = entry.get("command")            args = entry.get("args") or []            if not command:                raise RuntimeError(f"Server '{target}' is missing 'command'")            return [command, *args], child_env        if typ in ("sse", "http", "streamablehttp"):            url = entry.get("url")            if not url:                raise RuntimeError(f"Server '{target}' (type {typ}) is missing 'url'")            # Unified approach: always use current Python to run mcp-proxy module            cmd = [sys.executable, "-m", "mcp_proxy"]            if typ in ("http", "streamablehttp"):                cmd += ["--transport", "streamablehttp"]            # optional headers: {"Authorization": "Bearer xxx"}            headers = entry.get("headers") or {}            for hk, hv in headers.items():                cmd += ["-H", hk, str(hv)]            cmd.append(url)            return cmd, child_env        raise RuntimeError(f"Unsupported server type: {typ}")    # Fallback to script path (back-compat)    script_path = target    if not os.path.exists(script_path):        raise RuntimeError(            f"'{target}' is neither a configured server nor an existing script"        )    return [sys.executable, script_path], os.environ.copy()

export OPENCLAW_GATEWAY_TOKEN=你的真实token# 确保没有残留进程pkill -f xiaozhi_tunnel.pypkill -f openclaw_agent.pycd 你存储py文件的文件夹python3 xiaozhi_tunnel.py openclaw_agent.py

看到服务启动成功