第06章 Tool能力实现作者光谷老亢| 源码地址https://github.com/kang-airtc/mcp-mini-book数据层就绪后,本章把 5 个 Tool 的实现逐一接到 SQLite 工单库上。Tool 是 MCP 三大能力中最常用的一个,也是模型与外部世界发生交互的核心入口。5 个 Tool 涵盖三类典型用法:聚合统计(get_ticket_statistics)、单条详情(get_ticket_detail)、列表查询(query_tickets_by_status、query_tickets_by_type)、轻量分析(analyze_resolution_time)。三类用法在 SQL 写法、参数 schema、返回结构上各有差异,逐一展开后,读者将对 MCP Tool 的实现范式有完整认识。读完本章,读者将理解mcp.tool()装饰器的工作机制、参数 schema 的推导规则,以及 5 个 Tool 各自的 SQL 实现要点。6.1 Tool装饰器与签名约定mcp.tool()装饰器把普通 Python 函数提升为 MCP Tool。它做的事可以拆成三步:解析函数签名推导inputSchema、抽取 docstring 作为 description、把函数注册到 FastMCP 内部的工具表。最小调用形态如下:mcp.tool()defquery_tickets_by_status(status:str)-str:按状态查询工单 Args: status: 工单状态(open、in_progress、resolved、closed) Returns: JSON 格式的工单列表 ...# 业务实现框架从这个声明中提取三件事。第一,Tool 名取自函数名,即query_tickets_by_status,这一名字会出现在tools/list返回的清单中。第二,description 取 docstring,完整传递给 Client,再被转译为 Function Calling 的描述字段,直接影响模型的调用决策。第三,inputSchema由参数类型注解推导,status: str推导出{type: string},无默认值的参数被归入 required 列表;Args 块中的参数描述会被并入 schema 的description字段,作为模型决策时的辅助信息。返回值约定为字符串,内容通常是 JSON 序列化后的业务数据。Server 端把它包装为 content 数组中的 text 块返回给 Client;Client 拿到后通过json.loads还原。这一约定看似多走一步,实际上是为了在协议层保持简洁,所有 Tool 返回值都是 text 块,模型只需要解一种格式。完整的 Tool 调用链路如图 6-1 所示,从用户提问出发,经模型决策、Client 转发、Server 执行、结果回填,最终回到模型的下一轮推理。这条链路在任何 MCP 项目中都是统一的,Tool 的实现者只关心其中“Server 执行 SQL”这一环节。注意:docstring 是给模型看的,不是给读者看的。模型在决定是否调用某个 Tool 时,主要依据 description 的自然语言描述。如果 docstring 写得含糊,模型会出现“知道有这个 Tool 但不知该不该调”的状态,产生跳过或乱用的副作用。Args 块的参数描述应显式列出枚举取值范围,避免模型生成不合法参数。本章实现的 5 个 Tool 涵盖统计、详情、查询、分析四种用途,完整清单见表 6-1。表 6-1 工单分析服务的 5 个 Tool类别Tool 名参数返回用途统计get_ticket_statistics无JSON 统计概览提供按状态/类型/优先级的分布数据详情get_ticket_detailticket_no: strJSON 单条详情按工单号查具体工单查询query_tickets_by_statusstatus: strJSON 列表按状态筛选工单查询query_tickets_by_typeissue_type: strJSON 列表按问题类型筛选工单分析analyze_resolution_time无JSON 时长统计计算解决时间的均值与极值6.2 统计与详情类Toolget_ticket_statistics是 5 个 Tool 中最重的一个,但它对模型最友好,一次调用即可拿到工单池的全貌(总数、各状态分布、各类型分布、各优先级分布、平均满意度),省去模型多次串联调用的成本。先看函数前半段的查询部分:mcp.tool()defget_ticket_statistics()-str:获取工单统计概览connsqlite3.connect(DB_PATH)cursorconn.cursor()cursor.execute(SELECT status, COUNT(*) FROM tickets GROUP BY status)status_stats{row[0]:row[1]forrowincursor.fetchall()}第一段聚合按status分组计数,得到{open: 2, in_progress: 3, ...}这样的字典结构。同样的模式接下来分别用于issue_type与priority维度,最后把所有结果组装成一个 JSON:cursor.execute(SELECT issue_type, COUNT(*) FROM tickets GROUP BY issue_type)type_stats{row[0]:row[1]forrowincursor.fetchall()}cursor.execute(SELECT priority, COUNT(*) FROM tickets GROUP BY priority)priority_stats{row[0]:row[1]forrowincursor.fetchall()}conn.close()stats{总计工单数:sum(status_stats.values()),按状态分布:status_stats,按类型分布:type_stats,按优先级分布:priority_stats,}returnjson.dumps(stats,ensure_asciiFalse,indent2)关键设计有两点。其一,把多个独立维度的统计合到一次 Tool 调用中,模型一次拿到完整画像;若每个维度都独立成 Tool,模型为了组装总图会发起多轮调用,token 消耗成倍增加。其二,json.dumps显式传入ensure_asciiFalse保留中文,避免模型在后续推理中被\uXXXX转义序列干扰判断。get_ticket_detail通过ticket_no查询单条工单的完整字段,代码极简:mcp.tool()defget_ticket_detail(ticket_no:str)-str:获取单个工单的详细信息connsqlite3.connect(DB_PATH)conn.row_factorysqlite3.Row cursorconn.cursor()cursor.execute(SELECT * FROM tickets WHERE ticket_no ?,(ticket_no,))rowcursor.fetchone()conn.close()ifrow:returnjson.dumps(dict(row),ensure_asciiFalse,indent2)returnjson.dumps({error:f未找到工单{ticket_no}},ensure_asciiFalse)两个细节值得注意。其一,使用参数化占位符?防 SQL 注入;ticket_no来自模型生成的参数,理论上不可信,必须走 prepared statement。其二,设置conn.row_factory sqlite3.Row后,fetchone()返回的 row 支持按列名访问,dict(row)可直接转为带列名的字典,不必手工对照列序;查询不到时返回带error字段的 JSON,而非空对象,这样模型在拿到结果后能区分“查到但字段全空”与“压根没查到”两种语义。6.3 查询类Toolquery_tickets_by_status与query_tickets_by_type在结构上是对称的:同样按某一维度筛选、同样返回 JSON 列表、同样按created_at倒序。把两者并列展示,有助于读者识别 MCP Tool 在“列表查询”这一典型场景下的固定范式。先看query_tickets_by_status的签名与连接部分:mcp.tool()defquery_tickets_by_status(status:str)-str:按状态查询工单 Args: status: 工单状态(open、in_progress、resolved、closed) connsqlite3.connect(DB_PATH)conn.row_factorysqlite3.Row cursorconn.cursor()docstring 中status参数的取值列举写得很具体。这一信息会被 FastMCP 提取后透传给模型,模型在生成调用参数时会优先采用列表内的取值。如果省略这部分描述,模型可能传入“待处理”等中文取值,导致查询返回空。接下来是查询与返回部分:cursor.execute( SELECT ticket_no, customer_name, issue_type, priority, status, subject, created_at FROM tickets WHERE status ? ORDER BY created_at DESC ,(status,))rowscursor.fetchall()tickets[dict(row)forrowinrows]conn.close()returnjson.dumps(tickets,ensure_asciiFalse,indent2)SELECT 子句没有用SELECT *,而是显式列出 7 个字段。这是有意为之:列表场景通常不需要description这类大文本字段,显式列名既能减少传输体积,也避免把模型上下文塞满低密度信息。query_tickets_by_type的实现完全同构,只把WHERE status ?换成WHERE issue_type ?、把参数描述换成问题类型枚举,代码在此略去,完整版本见仓库agent-mcp-demo/mcp_server.py。6.4 分析类Tool:解决时间分布analyze_resolution_time演示了 Tool 能做的不止“查询”,还能承担轻量分析任务。它统计已解决工单的解决时长分布,返回平均、最大、最小三个指标。SQL 部分用 SQLite 内置的julianday函数把时间戳转换为天数:mcp.tool()defanalyze_resolution_time()-str:分析工单解决时间(仅统计已解决的工单)connsqlite3.connect(DB_PATH)cursorconn.cursor()cursor.execute( SELECT AVG(julianday(resolved_at) - julianday(created_at)) AS avg_days, MAX(julianday(resolved_at) - julianday(created_at)) AS max_days, MIN(julianday(resolved_at) - julianday(created_at)) AS min_days, COUNT(*) AS resolved_count FROM tickets WHERE resolved_at IS NOT NULL )两个工程要点值得记下。其一,在 SQL 层面通过WHERE resolved_at IS NOT NULL过滤未解决工单,而非把整表拉到 Python 端再过滤,这一写法对数据量增大后的扩展性至关重要。其二,julianday把时间戳转换为儒略日数(天为单位),做差后乘以 24 即可得到小时数,绕过了 Python 端复杂的日期解析。接下来是结果组装与容错处理:rowcursor.fetchone()conn.close()ifrowandrow[3]0:analysis{已解决工单数:row[3],平均解决时间:f{row[0]*24:.1f}小时,最长解决时间:f{row[1]*24:.1f}小时,最短解决时间:f{row[2]*24:.1f}小时,}else:analysis{message:暂无已解决的工单数据}returnjson.dumps(analysis,ensure_asciiFalse,indent2)代码有显式的容错分支:row[3](即resolved_count)大于 0 时返回统计结果,否则返回“暂无数据”提示。这种“业务上的空结果”应当作为正常返回值,而非异常向上抛,这一点对 MCP Tool 尤其重要。异常会被框架包装成 error 响应,模型在拿到 error 后可能误以为是技术故障而中止推理;而显式的message字段则能让模型理解“业务结果为空”并切换分析路径。5 个 Tool 实现完成后,Server 已具备核心动作能力。下一章把剩余的 2 个 Resource 与 2 个 Prompt 接上,完成 Server 的全部能力面。