多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

大模型Agent开发实战:用MCP协议搭建智能家居控制系统的完整配置流程

大模型Agent开发实战:用MCP协议搭建智能家居控制系统的完整配置流程 1. 智能家居 Agent 为什么总在“最后一公里”卡住大模型 Agent 能聊天、能写代码但一让它“下雨就关窗”很多人的项目就停在 Demo 阶段。问题不在模型智商而在模型和真实设备之间缺一层稳定的“接线板”。智能家居控制系统里设备协议五花八门有的走 HTTP有的走 MQTT有的只给你一个私有 SDK。你每接一个新设备就要在 Agent 里改一次工具描述、改一次参数解析改到最后代码里全是 if-else。MCP 协议Model Context Protocol模型上下文协议解决的正是这件事。它把“模型能调用的能力”抽象成标准原语——工具Tools、资源Resources、提示词Prompts用 JSON-RPC 2.0 在客户端和服务端之间通信。你可以把它理解成 USB-C设备端只要按 MCP 规范暴露工具Agent 端只要按 MCP 规范发现工具两边不用互相认识插上就能用。这篇面向已经写过基础 Agent、想跑通全链路的开发者。我会带你从零搭一个智能家居 MCP 服务端把“查天气”和“开关窗”封装成工具再写一个能自动选工具的 Agent 客户端最后用一句自然语言完成端到端控制测试。全程可复制配置片段直接拿去改。适合谁写过 Python、调过 OpenAI 或智谱类接口、知道什么是 function calling但还没把 MCP 真正跑起来的人。读完你能独立完成一次“模型自主决策 → 调用 MCP 工具 → 设备状态改变”的闭环。2. TaoToken 前置准备把模型调用这层先铺平在写 MCP 服务端之前先把模型调用这层铺平。Agent 要自动选工具必须有一个支持 function calling 的模型接口。我实测下来用 TaoToken 做统一入口比较省事它兼容 OpenAI 风格的调用格式MCP 客户端里拿到的工具描述可以直接塞进tools参数不用做二次转换。你需要准备三样东西Base URL、API Key、Model ID。这三件套在 MCP 客户端和 Agent 模块里都会用到建议先写进环境变量别硬编码。Base URL 用https://taotoken.net/api注意这个地址不带任何查询参数。API Key 在控制台的 API Keys 页面创建创建后只显示一次复制到本地.env文件里。Model ID 选一个支持工具调用的比如glm-4-plus这类具体以你账号下可用的为准。# .env 文件放在项目根目录 TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的实际key TAOTOKEN_MODEL_IDglm-4-plus如果你用的是 Claude Code 这类编码工具做 MCP 调试配置方式略有不同。Claude Code 的 MCP 配置走settings.json里面要写全 Base URL、Key、Model ID 三件套缺一个都会在初始化时报错。下面是一个可复制的片段路径按你本机的实际位置改{ mcpServers: { iot-server: { command: python, args: [/Users/you/project/server.py], env: { TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_API_KEY: sk-你的实际key, TAOTOKEN_MODEL_ID: glm-4-plus } } } }这里有个容易踩的坑env里的变量是传给 MCP 服务端进程的不是传给 Claude Code 本身的。如果你在服务端代码里用os.getenv(TAOTOKEN_API_KEY)读不到先检查是不是写错了层级。另外MCP 服务端的command和args必须指向真实存在的 Python 解释器和脚本路径相对路径在不同工作目录下会失效建议用绝对路径。把这三件套准备好之后后面所有模型调用都从这里取换模型只改一个环境变量不用动业务代码。这一步花五分钟能省掉后面反复改 key 的时间。3. 可复制配置MCP 服务端与 Agent 端完整片段这一节是全文的核心给你两份能直接跑的配置一份是 MCP 服务端把天气查询和窗户控制封装成工具一份是 Agent 端负责发现工具、让模型选工具、执行调用。先装依赖。MCP 的 Python 库叫mcp另外需要requests做 HTTP 请求cachetools做工具列表缓存python-dotenv读环境变量。pip install mcp requests cachetools python-dotenv服务端代码server.py如下。关键点是mcp.tool()装饰器它会自动解析函数名、参数名、类型注解和 docstring转成模型能理解的工具签名。docstring 一定要写清楚模型靠它判断什么时候该调这个工具。# server.py import os import requests from mcp.server.fastmcp import FastMCP mcp FastMCP(IoTServer) mcp.tool() def weather_query(location: str): 根据提供的城市名查询该城市当前天气情况 api_key os.getenv(WEATHER_API_KEY, your_amap_key) base_url https://restapi.amap.com/v3/weather/weatherInfo params {key: api_key, city: location} try: resp requests.get(base_url, paramsparams, timeout10) if resp.status_code 200: data resp.json() if data.get(lives): return data[lives][0] return {error: 无法获取天气信息请检查城市名} except Exception as e: return {error: f请求异常: {str(e)}} mcp.tool() def window_control(status: str): 控制窗户的开和关status 只能是 开 或 关 if status 开: return 窗户已打开 elif status 关: return 窗户已关闭 return f不支持的操作: {status} if __name__ __main__: mcp.run()Agent 端拆成两个模块。MCPClient负责连接服务端、缓存工具列表、执行工具调用MCPAgent负责跟模型对话、解析 tool_calls、循环执行直到任务完成。这种拆分的好处是换一个 MCP 服务端Agent 代码不用动。# agent.py import os import json import asyncio import cachetools from contextlib import AsyncExitStack from dotenv import load_dotenv from openai import OpenAI from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client load_dotenv() class MCPClient: def __init__(self): self.session None self.exit_stack AsyncExitStack() self.tools_cache cachetools.TTLCache(maxsize5, ttl5) self.CACHE_KEY tools async def connect_to_server(self, script_path: str): params StdioServerParameters( commandpython, args[script_path], envNone ) transport await self.exit_stack.enter_async_context(stdio_client(params)) self.stdio, self.write transport self.session await self.exit_stack.enter_async_context( ClientSession(self.stdio, self.write) ) await self.session.initialize() await self.list_tools() async def list_tools(self): if self.CACHE_KEY in self.tools_cache: return self.tools_cache[self.CACHE_KEY] resp await self.session.list_tools() tools [{ type: function, function: { name: t.name, description: t.description, parameters: t.inputSchema } } for t in resp.tools] self.tools_cache[self.CACHE_KEY] tools return tools async def call_tool(self, name, args): return await self.session.call_tool(name, args) async def cleanup(self): await self.exit_stack.aclose() class MCPAgent: def __init__(self, mcp_client): self.mcp_client mcp_client self.client OpenAI( base_urlos.getenv(TAOTOKEN_BASE_URL), api_keyos.getenv(TAOTOKEN_API_KEY) ) self.model os.getenv(TAOTOKEN_MODEL_ID) async def process_message(self, query: str) - str: messages [{role: user, content: query}] message await self.select_tools(messages) final_text [message.content or ] while message.tool_calls: for tc in message.tool_calls: name tc.function.name args json.loads(tc.function.arguments) result await self.mcp_client.call_tool(name, args) final_text.append(f[调用工具 {name}参数 {args}]) messages.append({ role: tool, tool_call_id: tc.id, content: str(result.content) }) message await self.select_tools(messages) final_text.append(message.content or ) return \n.join(final_text) async def select_tools(self, messages): tools await self.mcp_client.list_tools() resp self.client.chat.completions.create( modelself.model, messagesmessages, toolstools ) return resp.choices[0].message设备指令映射表建议单独维护别散在代码里。下面这张表是我实际项目里用的结构你可以直接扩展设备类型工具名参数取值底层动作窗户window_controlstatus开/关调用窗控 SDK空调ac_controltemp, mode16-30, cool/heat红外码发送灯光light_controlon, brightnesstrue/false, 0-100MQTT 发布把映射表独立出来后新增设备只需要加一行工具定义Agent 端完全无感。4. 验证请求从环境变量到工具调用的逐步动作配置写完别急着跑完整 Agent按顺序验证每一步出问题好定位。第一步验证环境变量读到了。在项目根目录执行python -c import os; from dotenv import load_dotenv; load_dotenv(); print(os.getenv(TAOTOKEN_BASE_URL), os.getenv(TAOTOKEN_MODEL_ID))输出应该是https://taotoken.net/api glm-4-plus。如果打印出None说明.env文件不在当前目录或者变量名拼错了。第二步单独验证 MCP 服务端能启动。直接运行python server.py正常情况下进程会挂起等待 STDIO 输入不报错就说明工具注册成功。如果报ModuleNotFoundError回去检查mcp库是否装在了当前 Python 环境。第三步验证工具发现。写一个最小测试脚本只连接服务端并列出工具# test_list.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters(commandpython, args[server.py], envNone) async with stdio_client(params) as (r, w): async with ClientSession(r, w) as session: await session.initialize() resp await session.list_tools() for t in resp.tools: print(t.name, -, t.description) asyncio.run(main())预期输出两行weather_query - 根据提供的城市名...和window_control - 控制窗户的开和关...。如果这里只看到一个工具检查服务端是不是有两个mcp.tool()装饰器。第四步验证模型能选对工具。跑完整 Agent# main.py import asyncio from agent import MCPClient, MCPAgent async def main(): client MCPClient() agent MCPAgent(client) try: await client.connect_to_server(./server.py) msg 查询今天深圳的天气。如果下雨就关窗否则开窗。 result await agent.process_message(msg) print(result) finally: await client.cleanup() asyncio.run(main())成功时你会看到类似输出[调用工具 weather_query参数 {location: 深圳}] [调用工具 window_control参数 {status: 开}] 今天深圳阴天气温 27℃没有下雨窗户已经打开了。注意执行顺序模型先调weather_query拿到结果后判断没下雨再调window_control传开。这个“两次调用”的循环正是 Agent 自主决策的体现。如果模型只调了一次就结束检查while message.tool_calls循环有没有写对以及工具返回结果有没有正确塞进messages。5. 本篇常见错排查401、local proxy failed 与 OAuth 报错跑不通的时候报错信息往往指向几个固定位置。我把高频错误和对应解法列出来你对照着查。401 Unauthorized。这个最常见九成是 API Key 问题。先确认.env里的TAOTOKEN_API_KEY没有多余空格再确认base_url写的是https://taotoken.net/api而不是带/v1的变体。有些客户端会自动补/v1补重了就会 401。如果 Key 是从控制台复制的注意别把前后引号也复制进去。local proxy failed / connection refused。这个报错通常出现在 MCP 客户端连接服务端时。STDIO 模式下客户端会用command和args启动一个子进程如果args里的脚本路径不对子进程起不来就会报连接失败。解法是把args改成绝对路径比如/Users/you/project/server.py。另外command要写python而不是python3还是反过来取决于你环境里哪个能跑通先用which python确认。reading choices 相关报错。这个一般出在模型返回结构解析上。如果你用的模型不支持 function callingresp.choices[0].message.tool_calls会是None后面while循环直接跳过看起来像“模型没选工具”。解法是换一个明确支持工具调用的 Model ID并在请求里确认tools参数确实传进去了。可以在select_tools里加一行print(len(tools))确认工具列表非空。OAuth 报错 / invalid_grant。如果你在 Claude Code 或类似工具里配置 MCP可能会遇到 OAuth 相关报错。这通常是因为工具本身走了 OAuth 流程而你的 MCP 服务端没实现对应的认证。对于本地 STDIO 模式一般不需要 OAuth检查是不是误配了 SSE 模式的远程地址。把配置改回commandargs的本地启动方式即可。工具调用结果为空。result.content在某些 MCP 版本里是列表直接str()会带一堆括号。建议在 Agent 里做一层提取content result.content[0].text if result.content else 。这样塞进messages的内容更干净模型下一轮判断也更准。排查顺序建议从下往上先确认服务端能单独启动再确认工具能列出再确认模型能选工具最后确认循环能收敛。每一步都有独立验证脚本别一上来就跑完整链路。6. 语义一致 CTA把这条链路接到你自己的设备上跑通天气开关窗只是起点。真正的智能家居控制系统里你要接的是真实设备灯、空调、窗帘、传感器。MCP 的价值在于你只需要在服务端新增一个mcp.tool()函数Agent 端一行都不用改模型就能自动发现并调用新能力。如果你想把模型调用这层统一管理方便换模型、看用量可以从 API Keys 页面创建新的 Key接入文档里有各语言的调用示例。调试阶段想先验证模型选工具的逻辑对不对可以直接在模型对话里试几轮确认工具描述写得够清楚。长期做编码和 Agent 开发的话Coding Plan 里包含了更完整的调用额度适合把这条链路跑成日常工具。下一步建议你做的是把window_control换成真实的设备 SDK 调用。比如接一个 MQTT 客户端把status参数映射成具体的 topic 和 payload。映射表已经在第 3 节给你了照着加一行就行。等你把灯和空调也接进去一句“有点热把空调开到 24 度顺便关下窗帘”就能触发三个工具的顺序调用——那时候你就真正拥有了一个能干活的家居 Agent。
返回列表