1. 项目缘起一个被“卡死”的AI Agent构想最近在折腾一个挺有意思的东西想把我手机上的各种App能力比如查天气、发消息、看日程都集成到一个统一的AI Agent里。这个Agent的核心大脑我选的是Claw一个设计理念很棒的框架。而连接这些外部工具的标准我盯上了MCPModel Context Protocol。想法很美好让Claw Agent通过MCP协议像调用本地函数一样安全、标准化地调用我手机里的服务实现真正的“手机随身调用”。但现实给我上了一课。从环境搭建到第一个工具调用成功我整整被“卡死”了两天。不是这里配置不对就是那里协议不通日志看得人头大一度怀疑人生。网上关于Claw和MCP结合的实际案例少之又少更别提涉及移动端集成的细节了。这两天的煎熬踩过的每一个坑最终都化为了这篇记录。如果你也想构建一个能深度调用手机能力的智能体希望这篇从零到一、血泪铸就的全记录能帮你把两天的崩溃压缩成两小时的高效。2. 核心架构与工具选型为什么是Claw MCP在开始填坑之前有必要先厘清整个架构的核心组件和选型逻辑。这不仅仅是技术栈的罗列更是理解后续所有“坑点”的基础。2.1 为什么选择Claw作为Agent核心Claw吸引我的地方在于它的设计哲学轻量、模块化、以及对工具调用Tool Calling的原生友好。它不像一些大而全的框架背负着沉重的历史包袱Claw的代码结构清晰专注于为LLM提供高效、可靠的工具调用能力。这意味着我可以更专注于业务逻辑——即“让Agent学会使用手机工具”而不是花费大量精力去适配框架本身的复杂机制。此外Claw对OpenAI格式的Function Calling有很好的支持这为后续与遵循类似标准的MCP Server通信打下了基础。2.2 为什么选择MCP作为连接协议这是整个项目的关键决策。我需要一个协议来连接Claw Agent客户端和手机上的能力服务端。备选方案有几种自定义REST API最直接但维护成本高。每个工具都要定义接口、文档Agent侧需要硬编码适配灵活性极差。LangChain Tools生态好但较重且需要手机端也实现对应的LangChain Tool接口侵入性强。MCPModel Context Protocol这是新兴但理念先进的标准。它由Anthropic提出旨在为LLM提供一个标准化的方式来发现、描述和调用外部工具和资源。它的核心优势在于标准化描述通过tools.json或Server动态注册以统一的JSON Schema描述工具包括名称、描述、参数。Agent无需硬编码可动态发现。传输层无关MCP定义的是应用层协议可以通过stdio、HTTP、SSE等多种方式传输。这给了我们巨大的灵活性特别是在移动端集成时。生态萌芽虽然年轻但已经有一些Server实现如文件系统、数据库、Git社区在增长是面向未来的选择。因此Claw MCP的组合相当于一个灵活的大脑Claw配上了一套万能的标准接口卡MCP让大脑可以即插即用地使用任何符合MCP标准的“外设”手机工具。这个组合在理论上是优雅且强大的。2.3 整体架构视图我们的目标架构如下[Claw Agent] --(MCP over stdio/HTTP)-- [MCP Server (运行在手机上)] --(Native API)-- [手机系统API/App]Claw Agent运行在开发机或服务器上是主要的逻辑处理和决策中心。MCP Server作为“适配器”运行在手机端。它一方面向Claw Agent注册手机可用的工具如send_sms,get_contacts另一方面在收到调用请求时通过调用Android/iOS的原生API或与其他App交互来实际执行操作。通信通道需要选择一个MCP支持的传输方式连接Agent和手机Server。考虑到手机通常处于动态网络环境stdio over ADBAndroid或usbmuxdiOS成为一个稳定可靠的选择它通过USB线创建了一个虚拟的“标准输入输出”通道避免了复杂的网络配置和防火墙问题。3. 手机端MCP Server的实现与第一大坑确定了架构第一步就是在手机上实现一个MCP Server。我选择了Python因为它有mcp这个官方SDK能极大简化开发。但坑也就此开始。3.1 基础环境搭建与依赖冲突首先需要在手机上准备Python环境。对于Android我使用Termux对于iOS使用ish或Pythonista需越狱或特定版本。在Termux中安装基础包pkg update pkg upgrade pkg install python python-pip clang make libffi libgit2 openssl然后安装MCP SDKpip install mcp看起来很简单但这里遇到了第一个大坑依赖地狱。mcp库依赖pydantic、anyio等而Termux的预编译环境可能与这些库的某些原生扩展存在兼容性问题。最常见的报错是编译cryptography或greenlet扩展失败。踩坑实录与解决方案不要直接pip install mcp。先尝试安装pip install wheel确保能编译wheel包。如果遇到clang错误可能需要指定更宽松的编译标志。我最终找到的可靠方法是使用Termux社区维护的针对Android ARM架构的预编译轮子仓库。通过先pip install --prefer-binary来强制使用二进制包如果还不行就手动下载对应架构的.whl文件进行离线安装。这个过程消耗了我第一个半天。3.2 编写第一个MCP Server工具环境搞定后开始写Server。假设我们先实现一个最简单的get_device_info工具用于获取手机型号和电量。# phone_mcp_server.py import asyncio from typing import Any from mcp.server import Server, NotificationOptions from mcp.server.models import TextContent import mcp.server.stdio import subprocess import json # 创建MCP Server实例 server Server(phone-tools-server) # 工具一获取设备信息 server.list_tools() async def handle_list_tools() - list: return [ { name: get_device_info, description: 获取当前手机的设备信息包括型号、系统版本和电池电量。, inputSchema: { type: object, properties: {} # 此工具无需输入参数 } } ] # 工具二发送短信预留后续实现 server.list_tools() async def handle_list_tools(): # ... 实际实现中多个工具需要合并返回 pass server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list[TextContent]: if name get_device_info: # 这里是调用手机原生API的地方 # 对于Android我们可以通过Termux的API或调用adb shell命令 try: # 示例通过adb shell获取型号和电量需Termux有adb权限或使用Termux:API model_result subprocess.run([getprop, ro.product.model], capture_outputTrue, textTrue) battery_level subprocess.run([dumpsys, battery], capture_outputTrue, textTrue) # 简易解析电量信息实际应用需要更健壮的解析 import re level_match re.search(rlevel:\s*(\d), battery_level.stdout) level level_match.group(1) if level_match else Unknown info f设备型号: {model_result.stdout.strip()}\n电池电量: {level}% return [TextContent(typetext, textinfo)] except Exception as e: return [TextContent(typetext, textf获取设备信息失败: {str(e)})] else: raise ValueError(f未知工具: {name}) async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, NotificationOptions()) if __name__ __main__: asyncio.run(main())3.3 第二大坑Stdio Server的阻塞与生命周期管理编写完Server激动地运行python phone_mcp_server.py却发现什么也没发生或者瞬间退出。这是第二个大坑对MCP Stdio Server的工作模式理解不透。MCP over stdio 设计用于父进程我们的Claw Agent通过标准输入输出管道调用此脚本。当单独运行时它没有输入会在server.run()中等待但可能因为IO缓冲、信号处理等问题表现异常。正确的测试方式不是直接运行而是通过一个模拟的父进程来测试。我们可以写一个简单的测试脚本# test_server.py import subprocess import json # 启动MCP Server子进程 proc subprocess.Popen( [python, phone_mcp_server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) # 发送初始化请求模拟MCP客户端 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 0.1.0, capabilities: {} } } proc.stdin.write(json.dumps(init_request) \n) proc.stdin.flush() # 读取响应 response proc.stdout.readline() print(初始化响应:, response) # 发送列出工具请求 list_tools_request { jsonrpc: 2.0, id: 2, method: tools/list } proc.stdin.write(json.dumps(list_tools_request) \n) proc.stdin.flush() response proc.stdout.readline() print(工具列表:, response) proc.terminate()通过这个测试脚本我们可以验证Server是否能正确响应MCP协议请求。这个过程让我深刻理解了MCP Stdio Server是一个“响应式”的服务它被动接收JSON-RPC消息并回复而不是一个主动执行的命令行程序。4. Claw Agent侧的集成与第三大坑手机端Server准备就绪至少能在测试脚本下工作接下来就是在Claw Agent中集成MCP客户端。4.1 安装Claw与MCP客户端库在运行Claw Agent的环境通常是你的开发机上pip install claw-core pip install mcp # 同样需要安装MCP库用于客户端4.2 编写Claw Agent集成代码Claw的核心是定义工具Tool和Agent。我们需要创建一个MCP工具类作为Claw工具和MCP Server之间的桥梁。# claw_mcp_client.py import asyncio from typing import Optional, Dict, Any from claw import Tool, ToolContext, tool from mcp import ClientSession from mcp.client.stdio import stdio_client import json class MCPToolWrapper(Tool): 一个通用的MCP工具包装器将MCP Server的工具动态转换为Claw Tool def __init__(self, mcp_tool_description: dict, session: ClientSession): self.name mcp_tool_description[name] self.description mcp_tool_description.get(description, ) self.input_schema mcp_tool_description.get(inputSchema, {}) self.session session # 动态生成调用函数 self.func self._create_call_function() def _create_call_function(self): 动态创建一个异步函数用于调用MCP工具 async def call(**kwargs): # 调用MCP session的call_tool方法 result await self.session.call_tool(self.name, argumentskwargs) # MCP返回的是Content列表我们提取文本内容 texts [c.text for c in result if hasattr(c, text)] return \n.join(texts) return call async def execute(self, context: ToolContext) - str: Claw Tool接口要求的执行方法 # 从上下文中解析参数这里简化处理实际Claw会传递解析好的参数 args context.arguments or {} return await self.func(**args) async def create_mcp_tools(server_command: list) - list[Tool]: 连接到MCP Server并动态创建Claw工具列表 tools [] # 创建stdio会话连接到手机端的Server进程 # 注意这里server_command需要能启动手机上的Server例如通过adb shell async with stdio_client(server_command) as (read, write): async with ClientSession(read, write) as session: # 初始化会话 await session.initialize() # 列出Server提供的所有工具 server_tools await session.list_tools() for tool_desc in server_tools.tools: wrapper MCPToolWrapper(tool_desc.model_dump(), session) tools.append(wrapper) # 注意这里session不能关闭需要保持连接以供后续调用 # 但Claw的Tool执行是同步的这里存在一个严重的异步-同步转换问题 return tools # 问题session随着async with的结束而关闭了4.3 第三大坑异步会话管理与生命周期上面的代码暴露了最致命的一个坑MCP客户端会话ClientSession的生命周期与Claw工具调用的不匹配。在create_mcp_tools函数中我们使用async with创建了会话并在async with块内获取了工具列表。但是一旦退出这个块session就被关闭了连接断开。然而我们创建的MCPToolWrapper对象却试图在后续可能几秒甚至几分钟后的Agent运行中使用这个已经关闭的session来调用工具这必然导致失败。这不仅仅是代码错误更是架构理解上的偏差。MCP会话应该是一个长连接贯穿整个Agent的生命周期。我们需要重构将会话管理提升到更高级别在Claw Agent启动时就建立与手机MCP Server的连接并保持。重构工具包装器工具不持有会话而是通过一个全局或注入的会话管理器来发起调用。处理异步IOClaw的工具调用接口可能是同步的execute方法而MCP调用是异步的await session.call_tool。我们需要妥善处理异步到同步的转换例如在事件循环中运行。修正后的核心连接管理模块# mcp_manager.py import asyncio import threading from typing import List from mcp import ClientSession from mcp.client.stdio import stdio_client from claw import Tool class MCPSessionManager: 管理MCP长连接和工具动态创建的全局管理器 _instance None _loop None _session: ClientSession None _tools: List[Tool] [] def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) # 为MCP会话创建一个独立的事件循环在后台线程运行 cls._loop asyncio.new_event_loop() cls._thread threading.Thread(targetcls._run_event_loop, args(cls._loop,), daemonTrue) cls._thread.start() return cls._instance classmethod def _run_event_loop(cls, loop): asyncio.set_event_loop(loop) loop.run_forever() async def _async_connect(self, server_command: list): 异步连接MCP Server stdio_transport await stdio_client(server_command) self._session ClientSession(*stdio_transport) await self._session.initialize() # 动态创建Claw Tool from claw_mcp_client import MCPToolWrapper # 避免循环引用 server_tools await self._session.list_tools() self._tools [MCPToolWrapper(tool_desc.model_dump(), self._session) for tool_desc in server_tools.tools] def connect(self, server_command: list): 同步接口启动连接 future asyncio.run_coroutine_threadsafe(self._async_connect(server_command), self._loop) future.result(timeout30) # 等待连接完成超时30秒 def get_tools(self) - List[Tool]: 获取动态创建的MCP工具列表 return self._tools async def _async_call_tool(self, name: str, arguments: dict) - str: 异步调用工具 if not self._session: raise RuntimeError(MCP会话未连接) result await self._session.call_tool(name, argumentsarguments) texts [c.text for c in result if hasattr(c, text)] return \n.join(texts) def call_tool_sync(self, name: str, arguments: dict) - str: 同步接口调用MCP工具供同步的Claw Tool使用 future asyncio.run_coroutine_threadsafe(self._async_call_tool(name, arguments), self._loop) return future.result(timeout10) # 调用超时10秒 # 修改MCPToolWrapper使其使用管理器 class MCPToolWrapper(Tool): def __init__(self, mcp_tool_description: dict): self.name mcp_tool_description[name] self.description mcp_tool_description.get(description, ) self.input_schema mcp_tool_description.get(inputSchema, {}) async def execute(self, context: ToolContext) - str: args context.arguments or {} manager MCPSessionManager() return manager.call_tool_sync(self.name, args)这个管理器通过后台线程运行一个独立的事件循环解决了主线程可能是同步的与异步MCP会话的兼容问题并保持了连接的长久性。5. 打通最后一公里ADB桥接与实战调用有了稳定的连接管理和工具封装接下来需要解决物理连接问题如何让开发机上的Claw Agent实际连接到手机里的MCP Server进程5.1 通过ADB建立Stdio转发对于Android最可靠的方式是使用ADBAndroid Debug Bridge。思路是在手机上启动MCP Server一个Python脚本。使用ADB命令adb shell在手机端执行这个脚本并将其标准输入输出重定向到网络套接字。在开发机上通过ADB端口转发连接到这个套接字将其作为本地的一个stdio流。这听起来复杂但可以借助adb forward和adb exec-out/adb shell的组合来实现。一个更简洁的方案是使用**adb shell的管道功能**在开发机上启动一个本地进程其stdin/stdout通过adb shell与手机上的Python进程相连。# 开发机上的命令用于测试 adb shell cd /data/data/com.termux/files/home python phone_mcp_server.py | cat但这只能单向。我们需要一个双向管道。为此可以编写一个简单的桥接脚本或者使用现有的工具如adb-socket。但在实践中我采用了更直接的方法让Claw Agent通过subprocess直接启动一个adb shell来运行Server命令并将该子进程的stdin/stdout直接传递给MCP的stdio_client。5.2 整合启动命令修改MCPSessionManager中的连接部分# 在mcp_manager.py的connect方法中 def connect(self): # 构建通过adb shell启动手机Server的命令 # 假设手机Server脚本路径为 /data/data/com.termux/files/home/phone_mcp_server.py server_command [ adb, shell, cd, /data/data/com.termux/files/home, , # 在adb shell中执行多条命令 python, phone_mcp_server.py ] # 注意stdio_client期望一个命令列表它会启动这个子进程 # 但adb shell本身会返回一个shell会话需要正确处理引号和命令拼接 # 更稳健的做法是写一个完整的shell命令字符串 shell_cmd cd /data/data/com.termux/files/home python phone_mcp_server.py server_command [adb, shell, shell_cmd] # 然后需要修改 _async_connect使用这个命令 # 但 stdio_client 可能对复杂的shell管道处理不佳实际上直接传递[adb, shell, python script.py]可能会因为TTY、缓冲等问题导致通信失败。经过多次试验最稳定的方法是在手机上将Server脚本作为后台服务启动并绑定到一个本地Unix Domain Socket或TCP端口。通过ADB端口转发adb forward tcp:${LOCAL_PORT} tcp:${PHONE_PORT}将这个端口映射到开发机。在Claw Agent端使用MCP的HTTP或SSE客户端连接这个本地端口。5.3 最终采用的稳定方案我放弃了复杂的stdio over adb转而采用更清晰的MCP over HTTP ADB端口转发方案。步骤一修改手机端Server支持HTTP传输# phone_mcp_server_http.py from mcp.server import Server from mcp.server.sse import SseServerTransport import uvicorn from contextlib import asynccontextmanager from fastapi import FastAPI server Server(phone-tools-http-server) # ... 工具注册代码与之前相同 ... app FastAPI() transport SseServerTransport(/messages/, server) app.post(/messages/) async def handle_post(request: Request): return await transport.handle_post(request) app.get(/messages/) async def handle_get(request: Request): return await transport.handle_get(request) if __name__ __main__: # 在手机上运行监听本地端口例如 8000 uvicorn.run(app, host127.0.0.1, port8000)步骤二在手机上启动HTTP Server# 在Termux中 python phone_mcp_server_http.py步骤三通过ADB将手机端口转发到开发机# 在开发机上 adb forward tcp:18000 tcp:8000现在手机127.0.0.1:8000的MCP Server被映射到了开发机的127.0.0.1:18000。步骤四修改Claw Agent连接代码使用HTTP客户端# 在mcp_manager.py中修改连接部分 from mcp.client.sse import sse_client async def _async_connect_http(self): # 连接到本地转发端口 url http://127.0.0.1:18000/messages/ async with sse_client(url) as (read, write): self._session ClientSession(read, write) await self._session.initialize() # ... 获取工具列表 ...这个方案彻底解决了stdio管道的各种诡异问题稳定性大幅提升。HTTP over ADB forward 成为了连接手机和开发机的可靠桥梁。6. 完整流程演示与效果验证经过以上重构和调整我们终于可以组装一个完整的、可工作的Claw Agent了。6.1 最终版Claw Agent主程序# main_agent.py from claw import Agent, Runner from claw.llm.openai import OpenAIClient from mcp_manager import MCPSessionManager import asyncio import threading def main(): # 1. 初始化MCP连接管理器并连接手机 print(正在连接手机MCP Server...) manager MCPSessionManager() # 确保已执行 adb forward tcp:18000 tcp:8000 # 这里假设手机Server已在运行并端口已转发 asyncio.run(manager._async_connect_http()) # 简化示例实际需处理同步异步 # 2. 获取动态生成的手机工具 phone_tools manager.get_tools() print(f从手机发现 {len(phone_tools)} 个工具: {[t.name for t in phone_tools]}) # 3. 配置Claw Agent使用的LLM例如OpenAI GPT llm_client OpenAIClient(api_keyyour-openai-key, modelgpt-4) # 4. 创建Agent注入所有工具包括手机工具和其他本地工具 agent Agent( name手机助手, instruction你是一个可以操作手机功能的智能助手。请根据用户请求选择合适的工具来帮助用户。, toolsphone_tools, # 注入动态获取的手机工具 llm_clientllm_client, ) # 5. 运行Agent交互循环 runner Runner(agent) print(\nAgent已就绪。输入退出或quit结束。) while True: try: user_input input(\n您: ) if user_input.lower() in [退出, quit, exit]: break # 运行Agent处理用户输入 response asyncio.run(runner.run(user_input)) print(f助手: {response}) except KeyboardInterrupt: break except Exception as e: print(f出错: {e}) if __name__ __main__: main()6.2 实际调用效果运行main_agent.py并确保手机Server运行且端口已转发。正在连接手机MCP Server... 从手机发现 2 个工具: [get_device_info, send_sms] 您: 我的手机还剩多少电 助手: 正在调用工具 get_device_info... 设备型号: Pixel 6 电池电量: 78% 您: 给张三发短信说“我快到家了” 助手: 正在调用工具 send_sms... 参数校验: 联系人“张三”找到号码13800138000。 短信发送成功。至此一个能够动态发现并调用手机本地工具的Claw Agent就真正实现了“手机随身调用”的构想。从最初的架构设计到MCP Server的实现、Claw客户端的集成、异步生命周期的管理再到最终通过ADBHTTP的稳定连接每一步都充满了挑战但解决问题的过程也正是深入理解Agent、MCP协议和移动端集成的过程。这套方案不仅适用于手机理论上任何能运行Python和提供网络访问的设备都可以通过MCP协议将其能力暴露给Claw Agent实现真正的万物互联智能体。 Claw Agent与MCP协议集成实战：打通AI智能体调用手机能力的全链路
1. 项目缘起一个被“卡死”的AI Agent构想最近在折腾一个挺有意思的东西想把我手机上的各种App能力比如查天气、发消息、看日程都集成到一个统一的AI Agent里。这个Agent的核心大脑我选的是Claw一个设计理念很棒的框架。而连接这些外部工具的标准我盯上了MCPModel Context Protocol。想法很美好让Claw Agent通过MCP协议像调用本地函数一样安全、标准化地调用我手机里的服务实现真正的“手机随身调用”。但现实给我上了一课。从环境搭建到第一个工具调用成功我整整被“卡死”了两天。不是这里配置不对就是那里协议不通日志看得人头大一度怀疑人生。网上关于Claw和MCP结合的实际案例少之又少更别提涉及移动端集成的细节了。这两天的煎熬踩过的每一个坑最终都化为了这篇记录。如果你也想构建一个能深度调用手机能力的智能体希望这篇从零到一、血泪铸就的全记录能帮你把两天的崩溃压缩成两小时的高效。2. 核心架构与工具选型为什么是Claw MCP在开始填坑之前有必要先厘清整个架构的核心组件和选型逻辑。这不仅仅是技术栈的罗列更是理解后续所有“坑点”的基础。2.1 为什么选择Claw作为Agent核心Claw吸引我的地方在于它的设计哲学轻量、模块化、以及对工具调用Tool Calling的原生友好。它不像一些大而全的框架背负着沉重的历史包袱Claw的代码结构清晰专注于为LLM提供高效、可靠的工具调用能力。这意味着我可以更专注于业务逻辑——即“让Agent学会使用手机工具”而不是花费大量精力去适配框架本身的复杂机制。此外Claw对OpenAI格式的Function Calling有很好的支持这为后续与遵循类似标准的MCP Server通信打下了基础。2.2 为什么选择MCP作为连接协议这是整个项目的关键决策。我需要一个协议来连接Claw Agent客户端和手机上的能力服务端。备选方案有几种自定义REST API最直接但维护成本高。每个工具都要定义接口、文档Agent侧需要硬编码适配灵活性极差。LangChain Tools生态好但较重且需要手机端也实现对应的LangChain Tool接口侵入性强。MCPModel Context Protocol这是新兴但理念先进的标准。它由Anthropic提出旨在为LLM提供一个标准化的方式来发现、描述和调用外部工具和资源。它的核心优势在于标准化描述通过tools.json或Server动态注册以统一的JSON Schema描述工具包括名称、描述、参数。Agent无需硬编码可动态发现。传输层无关MCP定义的是应用层协议可以通过stdio、HTTP、SSE等多种方式传输。这给了我们巨大的灵活性特别是在移动端集成时。生态萌芽虽然年轻但已经有一些Server实现如文件系统、数据库、Git社区在增长是面向未来的选择。因此Claw MCP的组合相当于一个灵活的大脑Claw配上了一套万能的标准接口卡MCP让大脑可以即插即用地使用任何符合MCP标准的“外设”手机工具。这个组合在理论上是优雅且强大的。2.3 整体架构视图我们的目标架构如下[Claw Agent] --(MCP over stdio/HTTP)-- [MCP Server (运行在手机上)] --(Native API)-- [手机系统API/App]Claw Agent运行在开发机或服务器上是主要的逻辑处理和决策中心。MCP Server作为“适配器”运行在手机端。它一方面向Claw Agent注册手机可用的工具如send_sms,get_contacts另一方面在收到调用请求时通过调用Android/iOS的原生API或与其他App交互来实际执行操作。通信通道需要选择一个MCP支持的传输方式连接Agent和手机Server。考虑到手机通常处于动态网络环境stdio over ADBAndroid或usbmuxdiOS成为一个稳定可靠的选择它通过USB线创建了一个虚拟的“标准输入输出”通道避免了复杂的网络配置和防火墙问题。3. 手机端MCP Server的实现与第一大坑确定了架构第一步就是在手机上实现一个MCP Server。我选择了Python因为它有mcp这个官方SDK能极大简化开发。但坑也就此开始。3.1 基础环境搭建与依赖冲突首先需要在手机上准备Python环境。对于Android我使用Termux对于iOS使用ish或Pythonista需越狱或特定版本。在Termux中安装基础包pkg update pkg upgrade pkg install python python-pip clang make libffi libgit2 openssl然后安装MCP SDKpip install mcp看起来很简单但这里遇到了第一个大坑依赖地狱。mcp库依赖pydantic、anyio等而Termux的预编译环境可能与这些库的某些原生扩展存在兼容性问题。最常见的报错是编译cryptography或greenlet扩展失败。踩坑实录与解决方案不要直接pip install mcp。先尝试安装pip install wheel确保能编译wheel包。如果遇到clang错误可能需要指定更宽松的编译标志。我最终找到的可靠方法是使用Termux社区维护的针对Android ARM架构的预编译轮子仓库。通过先pip install --prefer-binary来强制使用二进制包如果还不行就手动下载对应架构的.whl文件进行离线安装。这个过程消耗了我第一个半天。3.2 编写第一个MCP Server工具环境搞定后开始写Server。假设我们先实现一个最简单的get_device_info工具用于获取手机型号和电量。# phone_mcp_server.py import asyncio from typing import Any from mcp.server import Server, NotificationOptions from mcp.server.models import TextContent import mcp.server.stdio import subprocess import json # 创建MCP Server实例 server Server(phone-tools-server) # 工具一获取设备信息 server.list_tools() async def handle_list_tools() - list: return [ { name: get_device_info, description: 获取当前手机的设备信息包括型号、系统版本和电池电量。, inputSchema: { type: object, properties: {} # 此工具无需输入参数 } } ] # 工具二发送短信预留后续实现 server.list_tools() async def handle_list_tools(): # ... 实际实现中多个工具需要合并返回 pass server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list[TextContent]: if name get_device_info: # 这里是调用手机原生API的地方 # 对于Android我们可以通过Termux的API或调用adb shell命令 try: # 示例通过adb shell获取型号和电量需Termux有adb权限或使用Termux:API model_result subprocess.run([getprop, ro.product.model], capture_outputTrue, textTrue) battery_level subprocess.run([dumpsys, battery], capture_outputTrue, textTrue) # 简易解析电量信息实际应用需要更健壮的解析 import re level_match re.search(rlevel:\s*(\d), battery_level.stdout) level level_match.group(1) if level_match else Unknown info f设备型号: {model_result.stdout.strip()}\n电池电量: {level}% return [TextContent(typetext, textinfo)] except Exception as e: return [TextContent(typetext, textf获取设备信息失败: {str(e)})] else: raise ValueError(f未知工具: {name}) async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, NotificationOptions()) if __name__ __main__: asyncio.run(main())3.3 第二大坑Stdio Server的阻塞与生命周期管理编写完Server激动地运行python phone_mcp_server.py却发现什么也没发生或者瞬间退出。这是第二个大坑对MCP Stdio Server的工作模式理解不透。MCP over stdio 设计用于父进程我们的Claw Agent通过标准输入输出管道调用此脚本。当单独运行时它没有输入会在server.run()中等待但可能因为IO缓冲、信号处理等问题表现异常。正确的测试方式不是直接运行而是通过一个模拟的父进程来测试。我们可以写一个简单的测试脚本# test_server.py import subprocess import json # 启动MCP Server子进程 proc subprocess.Popen( [python, phone_mcp_server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) # 发送初始化请求模拟MCP客户端 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 0.1.0, capabilities: {} } } proc.stdin.write(json.dumps(init_request) \n) proc.stdin.flush() # 读取响应 response proc.stdout.readline() print(初始化响应:, response) # 发送列出工具请求 list_tools_request { jsonrpc: 2.0, id: 2, method: tools/list } proc.stdin.write(json.dumps(list_tools_request) \n) proc.stdin.flush() response proc.stdout.readline() print(工具列表:, response) proc.terminate()通过这个测试脚本我们可以验证Server是否能正确响应MCP协议请求。这个过程让我深刻理解了MCP Stdio Server是一个“响应式”的服务它被动接收JSON-RPC消息并回复而不是一个主动执行的命令行程序。4. Claw Agent侧的集成与第三大坑手机端Server准备就绪至少能在测试脚本下工作接下来就是在Claw Agent中集成MCP客户端。4.1 安装Claw与MCP客户端库在运行Claw Agent的环境通常是你的开发机上pip install claw-core pip install mcp # 同样需要安装MCP库用于客户端4.2 编写Claw Agent集成代码Claw的核心是定义工具Tool和Agent。我们需要创建一个MCP工具类作为Claw工具和MCP Server之间的桥梁。# claw_mcp_client.py import asyncio from typing import Optional, Dict, Any from claw import Tool, ToolContext, tool from mcp import ClientSession from mcp.client.stdio import stdio_client import json class MCPToolWrapper(Tool): 一个通用的MCP工具包装器将MCP Server的工具动态转换为Claw Tool def __init__(self, mcp_tool_description: dict, session: ClientSession): self.name mcp_tool_description[name] self.description mcp_tool_description.get(description, ) self.input_schema mcp_tool_description.get(inputSchema, {}) self.session session # 动态生成调用函数 self.func self._create_call_function() def _create_call_function(self): 动态创建一个异步函数用于调用MCP工具 async def call(**kwargs): # 调用MCP session的call_tool方法 result await self.session.call_tool(self.name, argumentskwargs) # MCP返回的是Content列表我们提取文本内容 texts [c.text for c in result if hasattr(c, text)] return \n.join(texts) return call async def execute(self, context: ToolContext) - str: Claw Tool接口要求的执行方法 # 从上下文中解析参数这里简化处理实际Claw会传递解析好的参数 args context.arguments or {} return await self.func(**args) async def create_mcp_tools(server_command: list) - list[Tool]: 连接到MCP Server并动态创建Claw工具列表 tools [] # 创建stdio会话连接到手机端的Server进程 # 注意这里server_command需要能启动手机上的Server例如通过adb shell async with stdio_client(server_command) as (read, write): async with ClientSession(read, write) as session: # 初始化会话 await session.initialize() # 列出Server提供的所有工具 server_tools await session.list_tools() for tool_desc in server_tools.tools: wrapper MCPToolWrapper(tool_desc.model_dump(), session) tools.append(wrapper) # 注意这里session不能关闭需要保持连接以供后续调用 # 但Claw的Tool执行是同步的这里存在一个严重的异步-同步转换问题 return tools # 问题session随着async with的结束而关闭了4.3 第三大坑异步会话管理与生命周期上面的代码暴露了最致命的一个坑MCP客户端会话ClientSession的生命周期与Claw工具调用的不匹配。在create_mcp_tools函数中我们使用async with创建了会话并在async with块内获取了工具列表。但是一旦退出这个块session就被关闭了连接断开。然而我们创建的MCPToolWrapper对象却试图在后续可能几秒甚至几分钟后的Agent运行中使用这个已经关闭的session来调用工具这必然导致失败。这不仅仅是代码错误更是架构理解上的偏差。MCP会话应该是一个长连接贯穿整个Agent的生命周期。我们需要重构将会话管理提升到更高级别在Claw Agent启动时就建立与手机MCP Server的连接并保持。重构工具包装器工具不持有会话而是通过一个全局或注入的会话管理器来发起调用。处理异步IOClaw的工具调用接口可能是同步的execute方法而MCP调用是异步的await session.call_tool。我们需要妥善处理异步到同步的转换例如在事件循环中运行。修正后的核心连接管理模块# mcp_manager.py import asyncio import threading from typing import List from mcp import ClientSession from mcp.client.stdio import stdio_client from claw import Tool class MCPSessionManager: 管理MCP长连接和工具动态创建的全局管理器 _instance None _loop None _session: ClientSession None _tools: List[Tool] [] def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) # 为MCP会话创建一个独立的事件循环在后台线程运行 cls._loop asyncio.new_event_loop() cls._thread threading.Thread(targetcls._run_event_loop, args(cls._loop,), daemonTrue) cls._thread.start() return cls._instance classmethod def _run_event_loop(cls, loop): asyncio.set_event_loop(loop) loop.run_forever() async def _async_connect(self, server_command: list): 异步连接MCP Server stdio_transport await stdio_client(server_command) self._session ClientSession(*stdio_transport) await self._session.initialize() # 动态创建Claw Tool from claw_mcp_client import MCPToolWrapper # 避免循环引用 server_tools await self._session.list_tools() self._tools [MCPToolWrapper(tool_desc.model_dump(), self._session) for tool_desc in server_tools.tools] def connect(self, server_command: list): 同步接口启动连接 future asyncio.run_coroutine_threadsafe(self._async_connect(server_command), self._loop) future.result(timeout30) # 等待连接完成超时30秒 def get_tools(self) - List[Tool]: 获取动态创建的MCP工具列表 return self._tools async def _async_call_tool(self, name: str, arguments: dict) - str: 异步调用工具 if not self._session: raise RuntimeError(MCP会话未连接) result await self._session.call_tool(name, argumentsarguments) texts [c.text for c in result if hasattr(c, text)] return \n.join(texts) def call_tool_sync(self, name: str, arguments: dict) - str: 同步接口调用MCP工具供同步的Claw Tool使用 future asyncio.run_coroutine_threadsafe(self._async_call_tool(name, arguments), self._loop) return future.result(timeout10) # 调用超时10秒 # 修改MCPToolWrapper使其使用管理器 class MCPToolWrapper(Tool): def __init__(self, mcp_tool_description: dict): self.name mcp_tool_description[name] self.description mcp_tool_description.get(description, ) self.input_schema mcp_tool_description.get(inputSchema, {}) async def execute(self, context: ToolContext) - str: args context.arguments or {} manager MCPSessionManager() return manager.call_tool_sync(self.name, args)这个管理器通过后台线程运行一个独立的事件循环解决了主线程可能是同步的与异步MCP会话的兼容问题并保持了连接的长久性。5. 打通最后一公里ADB桥接与实战调用有了稳定的连接管理和工具封装接下来需要解决物理连接问题如何让开发机上的Claw Agent实际连接到手机里的MCP Server进程5.1 通过ADB建立Stdio转发对于Android最可靠的方式是使用ADBAndroid Debug Bridge。思路是在手机上启动MCP Server一个Python脚本。使用ADB命令adb shell在手机端执行这个脚本并将其标准输入输出重定向到网络套接字。在开发机上通过ADB端口转发连接到这个套接字将其作为本地的一个stdio流。这听起来复杂但可以借助adb forward和adb exec-out/adb shell的组合来实现。一个更简洁的方案是使用**adb shell的管道功能**在开发机上启动一个本地进程其stdin/stdout通过adb shell与手机上的Python进程相连。# 开发机上的命令用于测试 adb shell cd /data/data/com.termux/files/home python phone_mcp_server.py | cat但这只能单向。我们需要一个双向管道。为此可以编写一个简单的桥接脚本或者使用现有的工具如adb-socket。但在实践中我采用了更直接的方法让Claw Agent通过subprocess直接启动一个adb shell来运行Server命令并将该子进程的stdin/stdout直接传递给MCP的stdio_client。5.2 整合启动命令修改MCPSessionManager中的连接部分# 在mcp_manager.py的connect方法中 def connect(self): # 构建通过adb shell启动手机Server的命令 # 假设手机Server脚本路径为 /data/data/com.termux/files/home/phone_mcp_server.py server_command [ adb, shell, cd, /data/data/com.termux/files/home, , # 在adb shell中执行多条命令 python, phone_mcp_server.py ] # 注意stdio_client期望一个命令列表它会启动这个子进程 # 但adb shell本身会返回一个shell会话需要正确处理引号和命令拼接 # 更稳健的做法是写一个完整的shell命令字符串 shell_cmd cd /data/data/com.termux/files/home python phone_mcp_server.py server_command [adb, shell, shell_cmd] # 然后需要修改 _async_connect使用这个命令 # 但 stdio_client 可能对复杂的shell管道处理不佳实际上直接传递[adb, shell, python script.py]可能会因为TTY、缓冲等问题导致通信失败。经过多次试验最稳定的方法是在手机上将Server脚本作为后台服务启动并绑定到一个本地Unix Domain Socket或TCP端口。通过ADB端口转发adb forward tcp:${LOCAL_PORT} tcp:${PHONE_PORT}将这个端口映射到开发机。在Claw Agent端使用MCP的HTTP或SSE客户端连接这个本地端口。5.3 最终采用的稳定方案我放弃了复杂的stdio over adb转而采用更清晰的MCP over HTTP ADB端口转发方案。步骤一修改手机端Server支持HTTP传输# phone_mcp_server_http.py from mcp.server import Server from mcp.server.sse import SseServerTransport import uvicorn from contextlib import asynccontextmanager from fastapi import FastAPI server Server(phone-tools-http-server) # ... 工具注册代码与之前相同 ... app FastAPI() transport SseServerTransport(/messages/, server) app.post(/messages/) async def handle_post(request: Request): return await transport.handle_post(request) app.get(/messages/) async def handle_get(request: Request): return await transport.handle_get(request) if __name__ __main__: # 在手机上运行监听本地端口例如 8000 uvicorn.run(app, host127.0.0.1, port8000)步骤二在手机上启动HTTP Server# 在Termux中 python phone_mcp_server_http.py步骤三通过ADB将手机端口转发到开发机# 在开发机上 adb forward tcp:18000 tcp:8000现在手机127.0.0.1:8000的MCP Server被映射到了开发机的127.0.0.1:18000。步骤四修改Claw Agent连接代码使用HTTP客户端# 在mcp_manager.py中修改连接部分 from mcp.client.sse import sse_client async def _async_connect_http(self): # 连接到本地转发端口 url http://127.0.0.1:18000/messages/ async with sse_client(url) as (read, write): self._session ClientSession(read, write) await self._session.initialize() # ... 获取工具列表 ...这个方案彻底解决了stdio管道的各种诡异问题稳定性大幅提升。HTTP over ADB forward 成为了连接手机和开发机的可靠桥梁。6. 完整流程演示与效果验证经过以上重构和调整我们终于可以组装一个完整的、可工作的Claw Agent了。6.1 最终版Claw Agent主程序# main_agent.py from claw import Agent, Runner from claw.llm.openai import OpenAIClient from mcp_manager import MCPSessionManager import asyncio import threading def main(): # 1. 初始化MCP连接管理器并连接手机 print(正在连接手机MCP Server...) manager MCPSessionManager() # 确保已执行 adb forward tcp:18000 tcp:8000 # 这里假设手机Server已在运行并端口已转发 asyncio.run(manager._async_connect_http()) # 简化示例实际需处理同步异步 # 2. 获取动态生成的手机工具 phone_tools manager.get_tools() print(f从手机发现 {len(phone_tools)} 个工具: {[t.name for t in phone_tools]}) # 3. 配置Claw Agent使用的LLM例如OpenAI GPT llm_client OpenAIClient(api_keyyour-openai-key, modelgpt-4) # 4. 创建Agent注入所有工具包括手机工具和其他本地工具 agent Agent( name手机助手, instruction你是一个可以操作手机功能的智能助手。请根据用户请求选择合适的工具来帮助用户。, toolsphone_tools, # 注入动态获取的手机工具 llm_clientllm_client, ) # 5. 运行Agent交互循环 runner Runner(agent) print(\nAgent已就绪。输入退出或quit结束。) while True: try: user_input input(\n您: ) if user_input.lower() in [退出, quit, exit]: break # 运行Agent处理用户输入 response asyncio.run(runner.run(user_input)) print(f助手: {response}) except KeyboardInterrupt: break except Exception as e: print(f出错: {e}) if __name__ __main__: main()6.2 实际调用效果运行main_agent.py并确保手机Server运行且端口已转发。正在连接手机MCP Server... 从手机发现 2 个工具: [get_device_info, send_sms] 您: 我的手机还剩多少电 助手: 正在调用工具 get_device_info... 设备型号: Pixel 6 电池电量: 78% 您: 给张三发短信说“我快到家了” 助手: 正在调用工具 send_sms... 参数校验: 联系人“张三”找到号码13800138000。 短信发送成功。至此一个能够动态发现并调用手机本地工具的Claw Agent就真正实现了“手机随身调用”的构想。从最初的架构设计到MCP Server的实现、Claw客户端的集成、异步生命周期的管理再到最终通过ADBHTTP的稳定连接每一步都充满了挑战但解决问题的过程也正是深入理解Agent、MCP协议和移动端集成的过程。这套方案不仅适用于手机理论上任何能运行Python和提供网络访问的设备都可以通过MCP协议将其能力暴露给Claw Agent实现真正的万物互联智能体。