1. 项目缘起：一个被“卡死”的AI Agent构想

最近在折腾一个挺有意思的东西，想把我手机上的各种App能力，比如查天气、发消息、看日程，都集成到一个统一的AI Agent里。这个Agent的核心大脑我选的是Claw，一个设计理念很棒的框架。而连接这些外部工具的标准，我盯上了MCP（Model 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（客户端）和手机上的能力（服务端）。备选方案有几种：

1. 自定义REST API：最直接，但维护成本高。每个工具都要定义接口、文档，Agent侧需要硬编码适配，灵活性极差。
2. LangChain Tools：生态好，但较重，且需要手机端也实现对应的LangChain Tool接口，侵入性强。
3. MCP（Model 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 ADB（Android）或usbmuxd（iOS）成为一个稳定可靠的选择，它通过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 SDK：

pip 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_output=True, text=True) battery_level = subprocess.run(['dumpsys', 'battery'], capture_output=True, text=True) # 简易解析电量信息（实际应用需要更健壮的解析） import re level_match = re.search(r'level:\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(type="text", text=info)] except Exception as e: return [TextContent(type="text", text=f"获取设备信息失败: {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'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # 发送初始化请求（模拟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, arguments=kwargs) # 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的生命周期。我们需要重构：

1. 将会话管理提升到更高级别：在Claw Agent启动时，就建立与手机MCP Server的连接并保持。
2. 重构工具包装器：工具不持有会话，而是通过一个全局或注入的会话管理器来发起调用。
3. 处理异步IO：Claw的工具调用接口可能是同步的（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(target=cls._run_event_loop, args=(cls._loop,), daemon=True) 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(timeout=30) # 等待连接完成，超时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, arguments=arguments) 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(timeout=10) # 调用超时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，最可靠的方式是使用ADB（Android Debug Bridge）。思路是：

1. 在手机上启动MCP Server（一个Python脚本）。
2. 使用ADB命令adb shell在手机端执行这个脚本，并将其标准输入输出重定向到网络套接字。
3. 在开发机上，通过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、缓冲等问题导致通信失败。经过多次试验，最稳定的方法是：

1. 在手机上将Server脚本作为后台服务启动，并绑定到一个本地Unix Domain Socket或TCP端口。
2. 通过ADB端口转发（adb forward tcp:${LOCAL_PORT} tcp:${PHONE_PORT}）将这个端口映射到开发机。
3. 在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, host="127.0.0.1", port=8000)

步骤二：在手机上启动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_key="your-openai-key", model="gpt-4") # 4. 创建Agent，注入所有工具（包括手机工具和其他本地工具） agent = Agent( name="手机助手", instruction="你是一个可以操作手机功能的智能助手。请根据用户请求，选择合适的工具来帮助用户。", tools=phone_tools, # 注入动态获取的手机工具 llm_client=llm_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客户端的集成、异步生命周期的管理，再到最终通过ADB+HTTP的稳定连接，每一步都充满了挑战，但解决问题的过程也正是深入理解Agent、MCP协议和移动端集成的过程。这套方案不仅适用于手机，理论上任何能运行Python和提供网络访问的设备，都可以通过MCP协议将其能力暴露给Claw Agent，实现真正的万物互联智能体。