ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

FastMCP框架进行MCP开发:(二)图书馆座位查询与预约MCP-Server

FastMCP框架进行MCP开发:(二)图书馆座位查询与预约MCP-Server 1. 从零跑通图书馆座位 MCP-ServerFastMCP 到底能做什么FastMCP 是一个基于 MCP 协议构建的快速开发框架它把网络通信、并发控制、工具注册这些底层细节都封装好了你只需要专注写业务逻辑。MCP 全称 Model Calling Protocol是一种模型与工具之间通信的标准协议允许模型根据任务需求动态调用外部工具比如 API、数据库等从而增强模型能力。这篇要做的是一个图书馆座位查询与预约的 MCP-ServerAI 客户端可以通过它查询某栋楼某天某个时间段还有哪些空座位也能直接发起预约、查看自己的预约记录。适合谁看如果你已经写过一点 Python想把自己的业务系统图书馆、会议室、工位、充电桩都行包装成 AI 能调用的工具这篇就是可跟做的模板。我会给出完整的server.py骨架、config.toml配置片段以及用 MCP 客户端发起查询和预约请求的验证动作。整个流程跑下来你会得到一个监听 8082 端口、支持 SSE 传输的可用服务。需要提前说明的是本文的后端 API 地址用的是示例域名你在本地调试时可以把它换成自己的 mock 服务或者真实接口。重点在于 FastMCP 这一层的工具定义、参数校验和调试方法这部分是通用的。2. TaoToken 前置准备把模型调用和 MCP 调试串起来MCP-Server 写完之后你需要一个能调用模型的客户端来验证工具是否被正确识别和触发。这里我用 TaoToken 来做模型侧的接入它的 API 地址是https://taotoken.net/api兼容常见的调用方式配置起来比较直接。第一步去控制台创建一个 API Key。打开https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_library_seat在 API Keys 页面新建一个密钥复制保存好。这个 Key 后面会用在客户端的配置里。第二步如果你打算长期做编码类或 Agent 类的开发可以了解一下 Coding Plan地址是https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_library_seat。它适合需要反复调试 MCP 工具、频繁发起模型请求的场景比单次调用更省心。第三步验证模型对话是否正常。打开https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_library_seat在对话框里发一句「你好帮我列一下你能调用的工具」确认模型能正常响应。这一步是为了排除 Key 或网络配置的问题等 MCP-Server 起来之后再回来测工具调用。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_library_seat里面有不同语言的调用示例遇到参数格式问题可以对照查。3. 可复制配置server.py 骨架与 config.toml先装依赖。FastMCP 的包名是mcp里面自带fastmcp模块pip install mcp requests然后创建server.py。下面这份代码可以直接复制运行我加了参数校验和错误处理比裸调用更稳import json import re import requests from datetime import datetime from typing import Optional from mcp.server.fastmcp import FastMCP mcp FastMCP( LibrarySeatService, descriptionf图书馆座位服务启动时间{datetime.now().strftime(%Y-%m-%d %H:%M)}, port8082, ) API_BASE http://127.0.0.1:9000/seat DATE_PATTERN re.compile(r^\d{4}-\d{2}-\d{2}$) TIME_RANGE_PATTERN re.compile(r^\d{2}:\d{2}~\d{2}:\d{2}$) def validate_date(date: str) - Optional[str]: if not DATE_PATTERN.match(date): return 日期格式错误应为 YYYY-MM-DD return None def validate_time_range(time_range: str) - Optional[str]: if not TIME_RANGE_PATTERN.match(time_range): return 时间段格式错误应为 HH:mm~HH:mm return None def format_response(response, success_keyNone): if response.status_code 200: data response.json() if success_key and success_key in data: return json.dumps({status: success, data: data}, ensure_asciiFalse) if isinstance(data, (list, dict)): return json.dumps({status: success, data: data}, ensure_asciiFalse) return json.dumps({status: error, message: 未找到有效数据}, ensure_asciiFalse) return json.dumps( {status: error, code: response.status_code, message: response.text}, ensure_asciiFalse, ) mcp.tool() async def query_available_seats( building: str, date: str, time_range: str, floor: Optional[str] None, seat_type: Optional[str] None, token: Optional[str] None, ) - str: 查询可用座位信息 err validate_date(date) or validate_time_range(time_range) if err: return json.dumps({status: error, message: err}, ensure_asciiFalse) params { building: building, date: date, timeRange: time_range, floor: floor, seatType: seat_type, } headers {Authorization: fBearer {token}, Content-Type: application/json} response requests.get(f{API_BASE}/available, headersheaders, paramsparams, timeout10) return format_response(response) mcp.tool() async def book_seat( seat_id: str, date: str, time_range: str, student_name: str, contact: str, token: Optional[str] None, ) - str: 预约指定座位 err validate_date(date) or validate_time_range(time_range) if err: return json.dumps({status: error, message: err}, ensure_asciiFalse) data { seatId: seat_id, date: date, timeRange: time_range, studentName: student_name, contact: contact, } headers {Authorization: fBearer {token}, Content-Type: application/json} response requests.post(f{API_BASE}/book, headersheaders, jsondata, timeout10) return format_response(response, success_keybookingId) mcp.tool() async def get_my_bookings(token: Optional[str] None) - str: 获取用户当前预约记录 headers {Authorization: fBearer {token}, Content-Type: application/json} response requests.get(f{API_BASE}/my-bookings, headersheaders, timeout10) return format_response(response) if __name__ __main__: mcp.run(transportsse)几个关键点解释一下。FastMCP初始化时传了port8082默认监听0.0.0.0本地调试没问题生产环境记得加鉴权。mcp.tool()装饰器把普通异步函数注册成 MCP 工具函数签名里的类型注解和 docstring 会被客户端读取用来判断什么时候调用这个工具。所以 docstring 要写清楚用途参数名要语义化。参数校验我单独抽了两个函数用正则检查日期和时间段格式。这一步很重要因为模型生成的参数不一定规范提前拦截能避免把脏数据打到后端。config.toml是给客户端用的配置片段放在客户端能读到的位置[mcp_servers.library_seat] url http://127.0.0.1:8082/sse transport sse description 图书馆座位查询与预约服务如果你的客户端支持 stdio 传输也可以改成command python、args [server.py]的形式但 SSE 更适合本地调试因为服务独立运行日志看得清楚。4. 验证请求从启动服务到成功预约先启动服务python server.py看到类似Uvicorn running on http://0.0.0.0:8082的输出就说明起来了。此时打开浏览器访问http://127.0.0.1:8082/sse应该能看到 SSE 流保持连接。接下来用 MCP 客户端验证。这里我用一个简单的 Python 客户端脚本模拟模型发起工具调用import asyncio from mcp import ClientSession from mcp.client.sse import sse_client async def main(): async with sse_client(http://127.0.0.1:8082/sse) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具, [t.name for t in tools.tools]) result await session.call_tool( query_available_seats, { building: A栋, date: 2025-06-15, time_range: 14:00~16:00, floor: 3楼, }, ) print(查询结果, result.content[0].text) asyncio.run(main())运行后你应该能看到三个工具名以及查询返回的 JSON。如果后端 mock 服务返回了座位列表输出里会有status: success和座位数据。预约的验证类似把call_tool换成book_seat传入seat_id、student_name、contact等参数。成功时返回里会带bookingId。我实测下来最容易出问题的是时间格式模型有时会生成14:00-16:00这种用短横线的写法被校验函数拦下来这时候在客户端提示里明确格式要求就能解决。如果你想在 TaoToken 的模型对话里直接测打开https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_library_seat把 MCP-Server 配置进去然后发一句「帮我查一下 A 栋 6 月 15 号下午两点到四点三楼的空座位」观察模型是否正确调用了query_available_seats。5. 本篇常见错排查报错一ModuleNotFoundError: No module named mcp说明依赖没装对。注意包名是mcp而不是fastmcp装完用pip show mcp确认版本。如果同时装了旧的fastmcp包可能会冲突建议先卸载再装。报错二Address already in use8082 端口被占了。用lsof -i :8082找到进程杀掉或者改FastMCP初始化时的port参数。改端口后记得同步改config.toml里的 URL。报错三工具列表为空客户端连上了但list_tools返回空。检查mcp.tool()装饰器是否加在函数上以及函数是否是async def。FastMCP 对同步函数也支持但异步函数在高并发下表现更好。另外确认mcp.run()在if __name__ __main__:里被调用。报错四调用工具返回status: error且 message 是连接错误后端 API 地址不通。API_BASE我写的是127.0.0.1:9000你需要有一个真实或 mock 的服务在跑。可以用python -m http.server 9000先起个静态服务测连通性或者用 FastAPI 写个最简单的 mock。报错五日期校验一直失败检查传入的日期是不是YYYY-MM-DD格式。模型有时会输出2025/06/15或6月15日这些都会被正则拦下。解决办法是在工具 docstring 里明确写「格式YYYY-MM-DD」模型看到后生成规范格式的概率会高很多。报错六SSE 连接建立后立即断开通常是客户端和服务端的传输协议不匹配。确认服务端mcp.run(transportsse)客户端用sse_client。如果客户端只支持 stdio就改用mcp.run(transportstdio)并调整配置。6. 下一步把座位服务接进你的编码工作流到这里一个能查询、能预约、能查记录的图书馆座位 MCP-Server 就跑通了。你可以把API_BASE换成真实的图书馆接口把参数校验按实际业务字段调整工具函数也可以继续加比如取消预约、修改时间段、查询历史记录。如果你打算把这个服务长期挂在本地配合编码助手一起用建议把 API Key 和 MCP 配置统一管理。API Keys 在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_library_seat创建和管理接入细节看https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_library_seat。需要长期跑 Agent 任务的话Coding Plan 在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_library_seat按需选用。最后留一个实用技巧调试 MCP 工具时先把format_response里的原始response.text打印出来确认后端返回结构再决定success_key怎么设。很多「工具调用失败」其实是后端字段名和预期不一致看一眼原始响应就能定位。
返回列表