快速手搓一个MCP服务指南(三):FastMCP的核心组件-构建MCP服务的关键技术实现
FastMCP 是一套面向 LLM 应用开发的工具框架,通过标准化协议衔接大语言模型与外部功能组件,构建「LLM+工具」的闭环交互体系。其核心技术体系包含四大模块:工具系统将 Python 函数转化为 LLM 可调用的能力单元,通过类型注解实现参数自动校验与协议转换;资源与模板机制提供数据访问接口,支持静态数据与参数化动态数据的标准化输出;提示系统通过参数化模板复用 LLM 交互逻辑,实现提示工程的模块化管理;上下文机制则贯穿请求生命周期,提供日志记录、进度反馈、LLM 二次调用等运行时能力。这套体系以「语义抽象+协议自动化」为核心设计理念,既屏蔽了底层通信细节,又通过标签过滤、动态禁用等机制实现细粒度的组件控制,最终帮助开发者高效构建类型安全、可扩展的 AI 应用系统,尤其适合需要多工具协同的复杂 LLM 场景。
FastMCP 工具核心技术文档详解
一、工具的本质与核心作用
FastMCP 工具是连接 LLM 与外部系统的桥梁,将 Python 函数转换为可被 LLM 调用的能力。其核心价值在于:
- 扩展 LLM 能力:让 LLM 能执行数据库查询、API 调用、文件操作等外部任务
- 类型安全交互:通过参数校验确保 LLM 输入符合函数预期
- 协议自动化:自动将函数转换为 MCP 协议支持的接口规范
当 LLM 调用工具时,流程如下:
- 发送带参数的请求 → 2. FastMCP 验证参数 → 3. 执行函数 → 4. 返回结果至 LLM
二、工具创建与装饰器用法
1. 基础工具定义
from fastmcp import FastMCPmcp = FastMCP(name="CalculatorServer")@mcp.tool
def add(a: int, b: int) -> int:"""两整数相加"""return a + b
- 工具名称默认使用函数名(
add
) - 描述默认使用函数文档字符串
- 自动生成参数 schema(基于类型注解)
2. 装饰器参数详解
参数名 | 类型 | 作用描述 |
---|---|---|
name | str | 自定义工具名称(覆盖函数名) |
description | str | 自定义描述(覆盖文档字符串) |
tags | set[str] | 工具标签(用于过滤或分类) |
enabled | bool | 初始化禁用工具(enabled=False ) |
exclude_args | list[str] | 隐藏参数(不暴露给 LLM) |
annotations | dict | 添加元数据(如只读标记、幂等性提示) |
示例:自定义工具元数据
@mcp.tool(name="find_products",description="按关键词搜索产品目录",tags={"search", "catalog"},annotations={"readOnlyHint": True}
)
def search(query: str, limit: int = 10) -> list:...
三、参数类型与校验
1. 类型注解的重要性
- LLM 引导:明确参数类型帮助 LLM 生成正确输入
- 自动校验:FastMCP 基于类型注解验证客户端输入
- 协议生成:生成符合 MCP 规范的 JSON Schema
2. 支持的参数类型
类型分类 | 示例注解 | 说明 |
---|---|---|
基础类型 | int , float , str , bool | 自动进行类型转换(如字符串 "42" 转为整数) |
集合类型 | list[float] , dict[str, int] | 支持嵌套结构(JSON 自动解析为对应类型) |
日期时间 | datetime , date | 接受 ISO 格式字符串(如 "2025-06-24" ) |
约束类型 | Literal["asc", "desc"] | 限制参数取值范围(LLM 只能输入指定值) |
Pydantic 模型 | class User(BaseModel) | 处理复杂结构化数据(自动校验嵌套字段) |
二进制数据 | bytes | 支持原始二进制或 Base64 编码(需手动解码) |
3. 参数元数据与校验
使用 pydantic.Field
增强参数定义:
from typing import Annotated
from pydantic import Field@mcp.tool
def process(text: Annotated[str, Field(min_length=3, max_length=500)],score: Annotated[float, Field(ge=0, le=100)],mode: Annotated[str, Field(pattern=r"^[a-z]+$")]
) -> str:...
min_length/max_length
:字符串长度限制ge/le/gt/lt
:数值范围约束pattern
:正则表达式校验description
:参数说明(展示给 LLM)
四、工具高级功能
1. 异步工具(IO 密集型任务)
import aiohttp
from fastmcp import FastMCPmcp = FastMCP()@mcp.tool
async def fetch_weather(city: str) -> dict:"""异步获取天气数据"""async with aiohttp.ClientSession() as session:resp = await session.get(f"https://api.weather.com/{city}")return await resp.json()
- 使用
async def
定义异步工具 - 适用于网络请求、数据库查询等耗时操作
- 避免阻塞服务器主线程
2. 工具返回值处理
FastMCP 自动转换返回值类型:
str
→ 文本内容dict/list
→ JSON 字符串bytes
→ Base64 编码(Blob 资源)Image/Audio
→ 专用内容类型(需导入fastmcp.utilities.types
)
示例:返回图片数据
from fastmcp.utilities.types import Image
from PIL import Image as PILImage
import io@mcp.tool
def generate_image(width: int, height: int) -> Image:img = PILImage.new("RGB", (width, height), "white")buffer = io.BytesIO()img.save(buffer, format="PNG")return Image(data=buffer.getvalue(), format="png")
3. 错误处理与安全控制
- 标准异常:抛出
ValueError
等常规异常,错误信息会返回给 LLM - ToolError:专用错误类(强制显示错误信息,不受
mask_error_details
影响) - 错误屏蔽:服务器配置
mask_error_details=True
隐藏详细堆栈
from fastmcp.exceptions import ToolError@mcp.tool
def divide(a: float, b: float) -> float:if b == 0:raise ToolError("除数不能为零")return a / b
五、MCP 上下文(Context)
工具可通过 Context
对象访问运行时功能:
from fastmcp import Context@mcp.tool
async def process_data(uri: str, ctx: Context) -> dict:# 日志记录await ctx.info(f"处理资源: {uri}")# 读取资源resource = await ctx.read_resource(uri)# 进度报告await ctx.report_progress(50, 100)# 调用 LLM 辅助summary = await ctx.sample("总结数据: " + resource[0].content)return {"summary": summary.text}
- 日志:
ctx.debug/info/warning/error
- 资源访问:
ctx.read_resource
- 进度跟踪:
ctx.report_progress
- LLM 交互:
ctx.sample
(在工具中调用 LLM)
六、工具管理与服务器行为
1. 标签过滤与禁用
- 标签过滤:通过服务器配置
include_tags
/exclude_tags
控制工具暴露 - 动态禁用:
tool.disable()
/tool.enable()
运行时控制工具状态
# 仅暴露带"public"标签的工具
mcp = FastMCP(include_tags={"public"})@mcp.tool(tags={"public"})
def public_tool():...@mcp.tool(tags={"internal"})
def internal_tool():... # 不会暴露
2. 重复工具处理
服务器创建时配置 on_duplicate_tools
策略:
"warn"
(默认):警告并替换"error"
:抛出异常阻止注册"replace"
:静默替换"ignore"
:忽略新注册
mcp = FastMCP(on_duplicate_tools="error")@mcp.tool
def my_tool():...@mcp.tool # 重复注册会抛出异常
def my_tool():...
3. 动态移除工具
mcp = FastMCP()@mcp.tool
def old_tool():...# 移除工具
mcp.remove_tool("old_tool")
七、关键概念总结
模块 | 核心要点 |
---|---|
工具定义 | @mcp.tool 装饰器,类型注解是关键,参数元数据用 Field 增强 |
异步处理 | async def 定义异步工具,适用于网络/IO 操作,避免阻塞 |
上下文交互 | Context 对象提供日志、资源访问、LLM 调用等功能 |
安全与控制 | 标签过滤、参数隐藏、错误屏蔽,ToolError 用于强制显示错误 |
服务器行为 | 重复工具策略、动态移除工具、legacy JSON 解析(默认禁用) |
八、实践建议
- 类型优先:始终添加类型注解,利用 Pydantic 校验提升鲁棒性
- 异步分离:IO 密集型任务使用异步工具,CPU 密集型用同步工具
- 元数据完善:通过
annotations
添加工具特性(只读/幂等),帮助客户端优化交互 - 错误防御:合理使用
ToolError
和mask_error_details
平衡调试与安全 - 标签管理:利用标签实现环境隔离(如开发环境暴露
debug
工具,生产环境隐藏)
通过上述功能,FastMCP 工具系统提供了从简单函数到复杂异步任务的完整解决方案,完美衔接 LLM 与外部系统的交互需求。
FastMCP 资源与模板技术文档详解
一、资源与模板的核心概念
FastMCP 资源(Resources)是客户端可读取的数据实体,而资源模板(Resource Templates)是动态生成资源的参数化定义。两者的核心作用是:
- 资源:提供静态或动态数据的只读访问(如配置文件、日志内容)
- 模板:通过 URI 参数动态生成资源(如按 ID 检索用户数据)
当客户端请求资源 URI 时,FastMCP 会:
- 匹配资源定义
- 执行对应的函数(如动态资源)
- 将结果转换为标准格式返回
二、资源定义与装饰器用法
1. 基础资源定义
from fastmcp import FastMCPmcp = FastMCP(name="DataServer")# 基础字符串资源
@mcp.resource("resource://greeting")
def get_greeting() -> str:"""返回问候语"""return "Hello from FastMCP Resources!"# JSON 格式资源(字典自动序列化)
@mcp.resource("data://config")
def get_config() -> dict:"""返回应用配置"""return {"theme": "dark", "version": "1.2.0"}
- URI:资源唯一标识符(如
resource://greeting
) - 延迟加载:仅在客户端请求时执行函数
- 元数据推断:函数名和文档字符串自动作为资源名称和描述
2. 装饰器参数详解
参数名 | 类型 | 作用描述 |
---|---|---|
uri | str | 资源唯一 URI(必填) |
name | str | 自定义资源名称(默认使用函数名) |
description | str | 自定义描述(默认使用文档字符串) |
mime_type | str | 内容类型(如 application/json ) |
tags | set[str] | 标签(用于客户端过滤) |
enabled | bool | 初始化禁用资源(enabled=False ) |
示例:完整资源定义
@mcp.resource(uri="data://app-status",name="ApplicationStatus",description="当前应用状态",mime_type="application/json",tags={"monitoring"}
)
def get_status() -> dict:return {"status": "ok", "uptime": 3600}
三、资源返回值处理
FastMCP 自动转换返回值为标准格式:
str
→ 文本资源(text/plain
)dict/list
→ JSON 文本(application/json
)bytes
→ Base64 编码(需指定mime_type
)None
→ 空内容
示例:返回二进制图片
from fastmcp.utilities.types import Image
import io
from PIL import Image as PILImage@mcp.resource("image://logo", mime_type="image/png")
def get_logo() -> bytes:img = PILImage.new("RGB", (200, 100), "blue")buffer = io.BytesIO()img.save(buffer, format="PNG")return buffer.getvalue()
四、资源高级功能
1. 动态启用/禁用资源(2.8.0+)
# 初始化禁用资源
@mcp.resource("data://secret", enabled=False)
def get_secret():return "敏感数据"# 运行时控制
config_resource = mcp.get_resource("data://config")
config_resource.disable() # 禁用资源
config_resource.enable() # 重新启用
2. 访问 MCP 上下文(2.2.5+)
from fastmcp import Context@mcp.resource("resource://system-info")
async def get_system_info(ctx: Context) -> dict:"""获取带请求 ID 的系统信息"""return {"status": "ok","request_id": ctx.request_id,"client_id": ctx.client_id}
- 上下文提供请求 ID、客户端 ID、日志记录等功能
3. 异步资源(处理 I/O 操作)
import aiofiles@mcp.resource("file:///app/logs/latest.log", mime_type="text/plain")
async def read_log() -> str:"""异步读取日志文件"""try:async with aiofiles.open("/app/logs/latest.log", "r") as f:return await f.read()except FileNotFoundError:return "日志文件未找到"
五、资源类直接注册
除装饰器外,可直接使用资源类注册静态资源:
from fastmcp.resources import FileResource, TextResource, DirectoryResource# 1. 注册文件资源
readme = FileResource(uri="file://README.md",path=Path("./README.md"),mime_type="text/markdown"
)
mcp.add_resource(readme)# 2. 注册文本资源
notice = TextResource(uri="resource://notice",text="系统维护将于周日进行",name="重要通知"
)
mcp.add_resource(notice)# 3. 注册目录列表资源
data_dir = DirectoryResource(uri="resource://data-files",path=Path("./data"),recursive=True # 包含子目录
)
mcp.add_resource(data_dir)
常见资源类:
TextResource
:纯文本内容FileResource
:本地文件HttpResource
:HTTP 远程资源DirectoryResource
:目录文件列表
六、资源模板(动态参数化资源)
1. 基础模板定义
@mcp.resource("weather://{city}/current")
def get_weather(city: str) -> dict:"""获取城市天气"""return {"city": city,"temp": 22,"condition": "晴朗"}
- URI 中的
{city}
为参数占位符 - 客户端请求
weather://london/current
时,city
参数被赋值为london
2. 多参数与默认值
@mcp.resource("search://{query}")
def search(query: str, limit: int = 10, sort: str = "relevance") -> dict:"""搜索资源(limit 和 sort 有默认值)"""return {"query": query,"limit": limit,"results": ["结果1", "结果2"]}
- 客户端可只提供
query
参数(如search://python
) - 可选参数使用默认值
3. 通配符参数(2.2.4+)
@mcp.resource("path://{filepath*}")
def get_path(filepath: str) -> str:"""匹配多级路径(如 path://docs/api/resources)"""return f"路径内容: {filepath}"@mcp.resource("repo://{owner}/{path*}/file.py")
def get_repo_file(owner: str, path: str) -> dict:"""匹配仓库中的文件路径"""return {"owner": owner,"file": f"{path}/file.py"}
{param*}
匹配多个路径段(含斜杠)- 适用于文件系统或层级数据访问
七、错误处理与服务器行为
1. 错误处理
from fastmcp.exceptions import ResourceError@mcp.resource("data://{id}")
def get_data(id: str) -> dict:if id == "invalid":raise ResourceError("数据 ID 无效")if id == "secret":raise ValueError("禁止访问敏感数据") # 会被 mask 处理return {"id": id, "data": "有效内容"}# 服务器配置错误屏蔽
mcp = FastMCP(mask_error_details=True)
ResourceError
的错误信息始终暴露- 其他异常在
mask_error_details=True
时会被简化
2. 重复资源处理
mcp = FastMCP(on_duplicate_resources="error") # 冲突时报错@mcp.resource("data://config")
def v1_config():return {"version": 1}@mcp.resource("data://config") # 重复注册会抛出异常
def v2_config():return {"version": 2}
- 策略包括:
"warn"
(默认,警告并替换)、"error"
(报错)、"replace"
(静默替换)、"ignore"
(忽略新注册)
八、核心功能总结
功能模块 | 核心特性 |
---|---|
资源定义 | @mcp.resource 装饰器,URI 唯一标识,自动类型转换 |
动态控制 | 启用/禁用、上下文访问、异步处理 |
资源类 | 预定义资源类型(文件、文本、目录),直接注册静态资源 |
模板参数 | URI 占位符、通配符参数、默认值处理,支持多级路径匹配 |
错误与冲突 | 自定义错误响应、错误信息屏蔽、重复资源策略配置 |
资源与模板是 FastMCP 中数据访问的核心组件,通过灵活的定义方式和参数化设计,使 LLM 能够动态获取所需数据,同时保持系统的可维护性和安全性。
FastMCP 提示(Prompts)技术文档详解
一、提示的核心概念与作用
FastMCP 提示(Prompts)是可复用的消息模板,用于指导 LLM 生成结构化响应。其核心价值在于:
- 标准化交互:定义统一的提示格式,确保 LLM 输出一致性
- 参数化设计:通过函数参数动态生成提示内容
- 复用性:跨场景共享提示逻辑,减少重复开发
当客户端调用提示时,流程如下:
- 匹配提示定义 → 2. 验证参数 → 3. 执行函数生成消息 → 4. 返回至 LLM 生成响应
二、提示定义与装饰器用法
1. 基础提示定义
from fastmcp import FastMCP
from fastmcp.prompts.prompt import Message, PromptMessage, TextContentmcp = FastMCP(name="PromptServer")# 基础字符串提示(自动转为用户消息)
@mcp.prompt
def ask_about_topic(topic: str) -> str:"""生成询问主题的用户消息"""return f"能否解释'{topic}'的概念?"# 自定义消息类型提示
@mcp.prompt
def generate_code_request(language: str, task_description: str) -> PromptMessage:"""生成代码请求消息"""content = f"编写{language}函数实现:{task_description}"return PromptMessage(role="user", content=TextContent(type="text", text=content))
- 名称:默认使用函数名(如
ask_about_topic
) - 参数:函数参数定义提示输入(如
topic: str
) - 元数据:文档字符串自动作为提示描述
2. 装饰器参数详解
参数名 | 类型 | 作用描述 |
---|---|---|
name | str | 自定义提示名称(覆盖函数名) |
description | str | 自定义描述(覆盖文档字符串) |
tags | set[str] | 标签(用于客户端过滤) |
enabled | bool | 初始化禁用提示(enabled=False ) |
示例:完整参数配置
@mcp.prompt(name="analyze_data",description="生成带参数的数据分析请求",tags={"data", "analysis"}
)
def data_analysis(data_uri: str = Field(description="数据资源URI"),analysis_type: str = Field(default="summary", description="分析类型")
) -> str:return f"请对{data_uri}的数据进行'{analysis_type}'分析"
三、参数类型与自动转换(2.9.0+ 新特性)
1. 类型注解支持
FastMCP 允许使用 Python 类型注解,自动处理 MCP 字符串参数的转换:
@mcp.prompt
def analyze_data(numbers: list[int], # 自动转换字符串为整数列表metadata: dict[str, str], # 转换JSON字符串为字典threshold: float # 转换字符串为浮点数
) -> str:avg = sum(numbers) / len(numbers)return f"平均值:{avg},超过阈值:{avg > threshold}"
2. 客户端调用格式
# 客户端需传递字符串格式的JSON
{"numbers": "[1, 2, 3, 4, 5]","metadata": "{\"source\":\"api\",\"version\":\"1.0\"}","threshold": "2.5"
}# 直接调用可使用原生类型(开发测试用)
result = await prompt.render({"numbers": [1, 2, 3],"metadata": {"source": "api"},"threshold": 2.5
})
四、返回值处理
FastMCP 支持多种返回类型,自动转换为 MCP 消息:
- str:转为单条用户消息
- PromptMessage:直接使用自定义消息
- list[PromptMessage/str]:转为消息序列(对话)
- 其他类型:尝试转为字符串后作为消息
示例:多消息返回
from fastmcp.prompts.prompt import Message@mcp.prompt
def roleplay_scenario(character: str, situation: str) -> list[Message]:"""生成角色扮演场景的初始对话"""return [Message(f"我们来角色扮演,你是{character},场景:{situation}"),Message("好的,我准备好了。接下来发生什么?", role="assistant")]
五、高级功能
1. 动态启用/禁用(2.8.0+)
# 初始化禁用
@mcp.prompt(enabled=False)
def experimental_prompt():return "这是实验性提示"# 运行时控制
holiday_prompt = mcp.get_prompt("seasonal_greeting")
holiday_prompt.disable() # 禁用提示
holiday_prompt.enable() # 重新启用
2. 异步提示(处理IO操作)
import aiohttp@mcp.prompt
async def data_driven_prompt(data_id: str) -> str:"""基于远程数据生成提示"""async with aiohttp.ClientSession() as session:resp = await session.get(f"https://api/data/{data_id}")data = await resp.json()return f"分析这些数据:{data['content']}"
3. 访问MCP上下文(2.2.5+)
from fastmcp import Context@mcp.prompt
async def generate_report(report_type: str, ctx: Context) -> str:"""生成带请求ID的报告请求"""return f"请创建{report_type}报告,请求ID:{ctx.request_id}"
六、服务器行为配置
1. 重复提示处理
mcp = FastMCP(on_duplicate_prompts="error") # 冲突时报错@mcp.prompt
def welcome():return "欢迎使用"@mcp.prompt # 重复注册会抛出异常
def welcome():return "您好"
- 策略包括:
"warn"
(默认,警告并替换)、"error"
(报错)、"replace"
(静默替换)、"ignore"
(忽略新注册)
七、核心功能总结
功能模块 | 核心特性 |
---|---|
定义方式 | @mcp.prompt 装饰器,支持同步/异步函数,类型注解增强 |
参数处理 | 自动转换字符串参数为原生类型,生成JSON schema指导LLM输入 |
返回值类型 | 支持字符串、消息对象、消息列表,自动转换为MCP标准格式 |
生命周期控制 | 动态启用/禁用,支持运行时状态切换 |
上下文交互 | 访问请求ID、客户端信息等运行时数据,增强提示动态性 |
提示(Prompts)是 FastMCP 中衔接 LLM 与应用逻辑的关键组件,通过参数化模板设计,既能保证交互标准化,又能灵活应对复杂场景,尤其适合需要多轮对话或结构化输出的 AI 应用开发。
FastMCP 上下文(Context)技术文档详解
一、Context 的核心概念与作用
FastMCP 上下文(Context
)是连接 MCP 功能与应用逻辑的桥梁,提供了在工具、资源、提示等组件中访问 MCP 核心能力的统一接口。其核心作用包括:
- 运行时交互:在函数执行过程中访问服务器功能(如日志、资源)
- 状态传递:携带请求元数据(如请求 ID、客户端标识)
- 能力扩展:支持 LLM 采样、进度报告等高级功能
Context
对象在请求处理期间自动注入,允许组件与 MCP 协议层深度交互,而无需关心底层实现细节。
二、Context 的获取方式
1. 依赖注入(推荐方式)
直接在函数参数中添加 Context
类型注解,FastMCP 会自动注入实例:
from fastmcp import FastMCP, Contextmcp = FastMCP(name="ContextDemo")@mcp.tool
async def process_file(file_uri: str, ctx: Context) -> str:"""处理文件,使用上下文进行日志和资源访问"""await ctx.info(f"开始处理文件: {file_uri}")return "文件处理完成"
2. 依赖函数(适用于嵌套调用)
通过 get_context()
函数在无法修改参数的场景中获取上下文:
from fastmcp.server.dependencies import get_contextasync def utility_function(data: list[float]) -> dict:ctx = get_context() # 仅在请求处理中有效await ctx.debug(f"处理 {len(data)} 个数据点")return {"processed": len(data)}@mcp.tool
async def analyze_data(dataset: str) -> dict:data = load_data(dataset)return await utility_function(data)
三、Context 的核心能力
1. 日志系统
提供分级日志功能,支持调试、信息、警告和错误日志:
@mcp.tool
async def calculate_average(numbers: list[float], ctx: Context) -> dict:await ctx.info(f"开始计算 {len(numbers)} 个数字的平均值")try:avg = sum(numbers) / len(numbers)await ctx.info(f"计算完成,平均值: {avg}")return {"average": avg}except ZeroDivisionError:await ctx.warning("接收到空数据列表")return {"error": "数据列表为空"}except Exception as e:await ctx.error(f"计算失败: {str(e)}")raise
日志方法列表:
ctx.debug(message)
:调试信息(开发阶段使用)ctx.info(message)
:正常执行信息ctx.warning(message)
:非致命警告ctx.error(message)
:错误信息ctx.log(level, message)
:自定义级别日志
2. 进度报告
长任务中向客户端反馈进度,提升用户体验:
@mcp.tool
async def process_batch(items: list[str], ctx: Context) -> dict:total = len(items)for i, item in enumerate(items):# 报告进度(0-100%)await ctx.report_progress(progress=i, total=total)await asyncio.sleep(0.1) # 模拟处理时间await ctx.report_progress(progress=total, total=total) # 完成return {"processed": total, "results": [i.upper() for i in items]}
关键参数:
progress
:当前进度值(如已处理数量)total
:总任务量(用于计算百分比)- 注意:客户端需在请求中提供
progressToken
才能接收进度更新
3. 资源访问
读取服务器注册的资源数据:
@mcp.tool
async def summarize_document(uri: str, ctx: Context) -> str:# 读取资源内容content_list = await ctx.read_resource(uri)if not content_list:return "文档为空"document = content_list[0].content# 简单摘要逻辑words = document.split()await ctx.info(f"文档包含 {len(words)} 个单词")if len(words) > 100:return f"摘要: {' '.join(words[:100])}..."return f"全文: {document}"
方法签名:
ctx.read_resource(uri: str) -> list[ReadResourceContents]
4. LLM 采样(2.0.0+ 新特性)
在工具中调用客户端的 LLM 进行文本生成:
@mcp.tool
async def analyze_sentiment(text: str, ctx: Context) -> dict:# 生成情感分析提示prompt = f"分析以下文本的情感(正面/负面/中性): {text}"# 调用客户端 LLMresponse = await ctx.sample(prompt,model_preferences="claude-3-sonnet", # 模型偏好temperature=0.7,max_tokens=50)# 处理结果sentiment = response.text.strip().lower()return {"text": text, "sentiment": sentiment}
高级用法:
@mcp.tool
async def generate_code(concept: str, ctx: Context) -> str:# 复杂提示(系统提示+用户消息)response = await ctx.sample(messages=["你是专业 Python 程序员,提供简洁可运行的代码示例",f"编写演示 '{concept}' 的 Python 代码示例"],temperature=0.5)return f"```python\n{response.text}\n```"
5. 请求信息获取
访问当前请求的元数据:
@mcp.tool
async def get_request_info(ctx: Context) -> dict:return {"request_id": ctx.request_id, # 请求唯一标识"client_id": ctx.client_id or "未知", # 客户端标识"session_id": ctx.session_id, # 会话 ID(HTTP 传输)}
四、高级访问能力
1. 服务器与会话访问
@mcp.tool
async def server_info(ctx: Context) -> str:# 访问服务器实例server_name = ctx.fastmcp.name# 底层会话访问(高级场景)session = ctx.sessionrequest_context = ctx.request_contextreturn f"服务器: {server_name}, 会话 ID: {session.session_id}"
2. HTTP 请求访问(2.2.7+ 特性)
@mcp.tool
async def http_request_info(ctx: Context) -> dict:# 获取 HTTP 请求对象(Web 场景)request = ctx.get_http_request()return {"user_agent": request.headers.get("user-agent", "未知"),"client_ip": request.client.host if request.client else "未知","path": request.url.path}
五、Context 的应用场景
- 调试与监控:在工具中记录关键操作日志,便于问题定位
- 长任务反馈:文件处理、数据导入等场景中实时显示进度
- 上下文感知:根据请求来源(客户端 ID)提供个性化响应
- LLM 协同:工具执行过程中调用 LLM 进行二次处理
- 资源联动:工具与资源数据动态交互(如读取配置后执行逻辑)
六、核心要点总结
功能模块 | 核心特性 |
---|---|
获取方式 | 依赖注入(参数注解)与依赖函数(get_context() ),自动生命周期管理 |
日志系统 | 四级日志分级,支持自定义日志级别,日志随请求上下文传递 |
进度报告 | 支持百分比进度显示,需客户端配合 progressToken |
LLM 采样 | 在工具中调用客户端 LLM,支持系统提示、模型偏好等高级参数 |
请求元数据 | 包含请求 ID、客户端 ID、会话 ID 等,用于追踪和权限控制 |
高级访问 | 服务器实例、HTTP 请求等底层访问,满足复杂场景扩展需求 |
Context
是 FastMCP 中实现组件与协议层交互的核心机制,通过它可以将工具、资源等功能与 MCP 的日志、进度、LLM 等能力无缝结合,构建更强大的 AI 应用系统。
相关文章:
快速手搓一个MCP服务指南(三):FastMCP的核心组件-构建MCP服务的关键技术实现
FastMCP 是一套面向 LLM 应用开发的工具框架,通过标准化协议衔接大语言模型与外部功能组件,构建「LLM工具」的闭环交互体系。其核心技术体系包含四大模块:工具系统将 Python 函数转化为 LLM 可调用的能力单元,通过类型注解实现参数…...
创建首个 Spring Boot 登录项目
📌 摘要 在 Java Web 开发中,登录功能是最基础也是最重要的模块之一。本文将手把手带你使用 IntelliJ IDEA 和 Maven 构建一个基于 Spring Boot 的简单登录系统,涵盖: 使用 IDEA 创建 Spring Boot 项目配置 Maven 依赖ÿ…...
order、sort、distribute和cluster by(Spark/Hive)
1. abstract ORDER BY:完整查询结果的全局行排序。与SORT BY、CLUSTER BY、DISTRIBUTE BY互斥,不能同时使用。 示例SELECT * FROM table_name ORDER BY column_name;SORT BY:只在每个分区内排序,局部排序结果不是全局有序。与ORD…...
# Python中等于号的使用
# Python中等于号的使用 ## 1. 问题的分析与思考 在Python中,等于号()是一个赋值运算符,用于将右侧的值或表达式的结果赋给左侧的变量。这是Python(以及许多其他编程语言)中非常基础且核心的一个概念。理…...
无人机神经网络模块运行与技术难点
一、神经网络模块的运行方式 1. 分层处理架构 感知层 多模态数据融合:通过八元数卷积网络(OCNN)统一处理LiDAR、摄像头、IMU等异构传感器数据,将点云坐标(x/y/z)、图像RGB与光流信息编码至8维虚部&#…...
宝塔服务器调优工具 1.1(Opcache优化)
第一步:宝塔服务器调优工具 1.1(按照下面的参数填写) 第二步:路径/www/server/php/80/etc/php.ini 搜索jit jit1235 其中1235根据服务器情况修改 第三步:路径/www/server/php/80/etc/php-cli.ini 搜索 jit1235 其中…...
day041-web集群架构搭建
文章目录 0. 老男孩思想-高薪四板斧1. web集群架构图2. 搭建异地备份服务2.1 服务端-阿里云服务器2.1.1 查看rsync软件包2.1.2 添加rsync配置文件2.1.3 添加虚拟用户2.1.4 创建校验用户密码文件2.1.5 创建备份目录2.1.6 启动服务2.1.7 开放安全组端口2.1.8 发送检查邮件 2.2 客…...
国产化条码类库Spire.Barcode教程:如何使用 C# 读取 PDF 中的条码(两种方法轻松实现)
在 PDF 文档的 .NET 平台处理流程中,使用 C# 读取 PDF 条码 是一项常见需求,特别适用于处理扫描件或电子表单。无论是物流、金融、医疗还是制造行业,PDF 文档中经常包含用于追踪或识别的条码。这些条码可能是嵌入图像,也可能是矢量…...
vue 3 计算器
效果: <template><div class"calculator-container"><div class"calculator"><!-- 显示区域 --><div class"display">{{ formattedDisplay }}</div><!-- 按钮区域 --><div class"…...
CRMEB PHP多门店版v3.2.1系统全开源+Uniapp前端+搭建教程
一.介绍 CRMEB多店版是一款为品牌连锁门店打造的私域电商解决方案,以三大运营模式为核心,助力品牌连锁门店轻松构建全渠道、一体化的私域电商生态,促进“线上电商”与“线下门店”销售运营融合,加速品牌数字化转型,为…...
主机复制文字和文件到 Ubuntu 虚拟机
在 VMware Workstation Pro 16 中复制文字和文件到 Ubuntu 虚拟机,方法如下: Open-VM-Tools 禁用 Wayland 解决 。 1.安装 VMware Tools(推荐)或 open-vm-tools: sudo apt update sudo apt install open-vm-tools…...
性能测试 —— 数据库的连接池和主从同步和分表分区
一、数据库的调优(库层面) 1、数据库连接池 1、介绍:数据库连接池(Database Connection Pool)是一种用于管理数据库连接的技术,它通过预先创建并维护一组数据库连接来提高应用程序的性能和可扩展性。 2、创建、管理、关闭 数据…...
猿人学js逆向比赛第一届第十二题
一、分析请求 看到这里只有一个m的密文参数,没有cookie,请求头等其他的参数,那么这里跟一堆栈信息。 很顺利地锁定了m的加密位置。看到是字符串拼接然后使用btoa函数进行编码,那么这里尝试使用Python复现一下。顺利拿到结果。 复现…...
第十节 新特性与趋势-CSS层叠规则升级
以下是关于 CSS层叠规则升级 的全面解析,结合最新规范(如级联层layer)和传统层叠机制的演进,从核心原理、应用场景到实践策略的系统性总结: 一、传统层叠规则的三大支柱 CSS层叠规则的传统机制基于以下三个维…...
关键领域软件工厂的安全中枢如何全面升级供应链检测能力
随着软件供应链安全体系在互联网、金融等领域逐步成熟,关键领域正加速迈向以 MLOps、软件工厂为核心的新型研发生态。在这一过程中,面对代码安全、依赖合规、系统可信等多重挑战,传统人工审查模式已难以满足国家级高安全性要求。 Gitee Scan…...
西门子G120XA变频器:数据中心能效革命的核心引擎
在数字经济爆发式增长的今天,数据中心已成为支撑社会运转的"数字心脏"。然而,其庞大的能耗需求与绿色低碳目标之间的矛盾日益凸显——尤其是冷却系统作为数据中心第二大能耗单元(占比约35%),正成为能效提升的…...
从零开始学习Spring Cloud Alibaba (一)
人狠话不多,直接开始少点屁话本着共同学习进步的目的和大家交流如有不对的地方望铁子们多多谅解 准备工具 开发工具 idea Java环境 jdk17 容器: docker Maven 3.8.6 仓库镜像阿里云 <mirror><id>alimaven</id><name>aliyun maven</name><…...
【C/C++】C++ 编程规范:101条规则准则与最佳实践
C 编程规范:101条规则准则与最佳实践 引言 C 是一门强大而复杂的语言,能高效控制硬件,也能写出优雅抽象。然而,正因其复杂性,项目中若缺乏统一规范,极易陷入混乱、难维护、易出错的泥潭。 本文总结了 10…...
PyTorch topk() 用法详解:取最大值
torch.topk(input, k) 返回张量中最大的 k 个元素以及它们在原张量中的 索引。 函数原型 torch.topk(input, k, dimNone, largestTrue, sortedTrue)参数说明: 参数说明input输入张量k要取出的前 k 个值dim指定沿哪个维度取值(默认是最后一维ÿ…...
毕业论文查重原理及降重方法
【30%的重复率有那么重要吗?】 老师说论文重读率必须低于30%,否则无法毕业! 如果您在专科或者普通本科学生,我只能这样一句话告诉你:你想多了~, 真的想多了~~~,一篇论文还不至于让你不能毕…...
Golang Kratos 系列:业务分层的若干思考(二)
上一篇文章简单讨论了领域层在Kratos中的使用,主要涉及引入领域层,将数据层和业务层之间的解耦,接下来讨论一个稍微全面一点的例子,在此基础上引入外部Api(主要是易变部分)的领域层下的情况。 我们同样可以…...
技术伦理之争:OpenAI陷抄袭风波,法院强制下架宣传视频
在AI巨头OpenAI宣布以65亿美元天价收购苹果前设计总监Jony Ive的硬件公司IO仅一个月后,一场抄袭指控将这家科技明星企业推上风口浪尖。 源自谷歌X实验室的初创企业IYO将OpenAI告上法庭,指控其窃取智能耳塞核心技术,并通过巨额收购试图掩盖抄袭…...
烟花爆竹生产企业库房存储安全风险预警系统
烟花爆竹生产企业库房存储安全风险预警系统是保障库房物资安全、规范作业流程、防范安全事故的重要技术手段,涵盖多个关键预警功能。 温湿度预警 在库房内安装温湿度传感器,这些传感器如同敏锐的“环境感知员”,能够实时监测库房内环境变…...
Jenkins+Jmeter+Ant接口持续集成
2025最新Jmeter接口测试从入门到精通(全套项目实战教程) 前言: 为什么要用Jmeter做接口测试: 当选择这套方案的时候,很多人会问,为什么选择Jmeter做Case管理?为什么不自己写框架?说…...
基于STM32的寻迹小车设计
标题:基于STM32的寻迹小车设计 内容:1.摘要 本文围绕基于STM32的寻迹小车设计展开。背景是随着自动化技术的发展,寻迹小车在工业巡检、物流运输等领域有广泛应用前景。目的是设计一款能稳定、准确寻迹的小车。方法上,以STM32微控制器为核心,…...
【150】基于SSM+Vue实现的小说阅读小程序(有文档)
系统介绍 基于SSMVue实现的小说阅读小程序采用前后端分离的架构方式,系统设计了管理员、用户两种角色,系统分为管理端、小程序端,管理端实现了管理员登录、个人中心、管理员管理、帮助中心管理、基础数据管理、论坛中心管理、公告资讯管理、…...
ValKey中使用SIMD指令优化bitcount命令
一、AVX/AVX2 的历史演进 随着计算机技术的飞速发展,数据处理需求呈指数级增长,SIMD(单指令多数据)技术应运而生。它通过一条指令同时处理多个数据元素,大幅提升计算效率,从早期的 MMX 技术起步࿰…...
leetcode114-二叉树展开为链表
leetcode 114 思路 用简单例子推导规律 不要一开始就看复杂的树,先从最简单的情况入手 案例一:只有一个节点 输入:1 输出:1不需要任何操作,直接返回 案例二:有两个节点 输入: 1/2输出&a…...
第七章 习题
1.给出下面表达式的逆波兰表示(后缀式): 3请将表达式-(ab)* (cd)-(abc)分别表示成三元式,间接三元式和四元式序列 四元式(Op,arg1,arg2,result) (,a,b,T1) (,c,d,T2) (*,T1,T2,T3) (uminus,T3,-,T4) (,a,b,T5) (,T5,c,T6) (-,T4,T6,r) 三元式 (op,arg1,arg2) (0) (,…...
Spring Ai Alibaba Graph实现五大工作流模式
Spring Ai Alibaba Graph实现五大工作流模式 概述 在 building-effective-agents 一文中,Anthropic将"智能体系统"(agentic systems),从架构层面分为 “工作流”(workflows)和 “智能体”(agents): 工作流…...
基于单片机的语音控制设计(论文)
摘要 自然语音作为人机交互在目前得以广泛的应用以及极大的发展前景。该设计介绍了基于非指定人语音芯片LD3320的语音控制器结构及其实现语音控制的方法。该语音控制器利用STM32F103C8T6单片机作为主要控制器,控制芯片对输入的进行语音识别并处理,根据语…...
【网络安全】从IP头部看网络通信:IPv4、IPv6与抓包工具 Wireshark 实战
从IP头部看网络通信:IPv4、IPv6与抓包工具 Wireshark实战 在网络安全分析和数据通信的世界中,一切都始于“数据包”。数据包是网络上传输的基本单位,而数据包的结构与内容,正是我们理解网络行为的核心。本文将带你深入了解 IP 协…...
计算机组成原理笔记(公众号版本)
1.MAR,存储单元和PC位数之间的关系 我们的MAR位数取决于我们的存储单元的个数; PC位数也是取决于我们的存储单元的个数; 假设我们是64个存储单元,我们的这个MAR实际上就是6位,因为这个2的6次方等于我们的64吗&#…...
CPM基本原理
CPM(连续相位调制,Continuous Phase Modulation )是一种恒包络,信号相位随时间连续变化的调制技术 。其优势显著,连续相位特性,频谱效率高,对带外辐射抑制好;相位平滑过渡࿰…...
浅谈开源在线客服系统与 APP 集成的技术方案与优劣势
在为移动端 App 接入在线客服系统的过程中,我经历了长时间的技术选型探索。最初,我也曾被一些“技术理想主义”选项所吸引,比如让用户自己研发界面我提供 API 以获得最高自由度,或集成 SDK 以追求原生体验。然而,随着项…...
AutoGPT,自主完成复杂任务
AutoGPT是一个开源的AI Agent项目,它的核心目标是让AI能够自主完成复杂任务,而不仅仅是回答单个问题。简单来说,它让AI具备了"自主思考和行动"的能力。 1. AutoGPT的核心概念 什么是AI Agent? AI Agent(智…...
基于Qt C++的影像重采样批处理工具设计与实现
摘要 本文介绍了一种基于Qt C++框架开发的高效影像重采样批处理工具。该工具支持按分辨率(DPI) 和按缩放倍率两种重采样模式,提供多种插值算法选择,具备强大的批量处理能力和直观的用户界面。工具实现了影像处理的自动化流程,显著提高了图像处理效率,特别适用于遥感影像处…...
Qt Windows平台调用ffmpeg动态库
本文基于QT6.8实测验证。 一、下载预编译库 从官方或第三方源获取FFmpeg的Windows动态库(Shared版本),解压后需包含以下目录: bin DLL文件(运行时依赖)include 头文件lib .lib或.dll.a链接…...
猿人学js逆向比赛第一届第十三题
一、分析请求 通过分析请求得知,本题目的参数为yuanrenxue_cookie,因为题目中中明说了是动态cookie,所以可以先删除这个cookie然后下脚本断点,简单分析看一下页面执行的js代码逻辑。 在经过这段代码的时候很明显的可以看到这里的有…...
React性能优化精髓之一:频繁setState导致滚动卡顿的解决方案
在开发一个 List 页面时,我们遇到了一个典型的React性能问题:页面在滚动时出现明显卡顿。这个问题的调试过程充满了误判和重新思考,最终发现了一个重要的性能优化原则。 问题现象 我们有一个监控仪表盘页面,包含多个图表组件。用…...
JavaScript 事件常用属性
一、事件对象基础 在事件处理函数中,浏览器会自动传入一个 event 对象,它包含了与事件相关的所有信息。 element.addEventListener(click, function(event) {// event 是事件对象 }); 二、常见事件属性 属性名 类型 描述 type String 事件类…...
配置自己的NTP 服务器做时间同步
✅ 推荐方案:使用 chrony 搭建 NTP 服务器(适用于 CentOS 7/8/9) chrony 是 CentOS 推荐的 NTP 实现,精度高、资源占用低、同步快,默认在 CentOS 8 中取代了 ntpd。 🔧 一、安装 chrony sudo yum install…...
基于深度学习的双色球智能预测系统:从原理到实现
需要源码的小伙伴可以在这里直接下载:基于深度学习的双色球智能预测系统:从原理到实现(完整代码训练数据)可直接运行-预测)资源-CSDN下载可直接运行,包括完整的训练测试数据,让你的双色球更准&a…...
STM32[笔记]--4.嵌入式硬件基础
4.嵌入式硬件基础 4.1认识上官二号开发板 主控芯片:STM32F103C8T6高速晶振:8M低速晶振:32.768kLED:5颗KEY:3个 主控芯片内部的资源如下项目介绍内核Cortex-M3Flsah64K*8bitSRAM20K*8bitGPIO37个GPIO,分别为PA0-PB15,PC13-PC15,PD0-PD1ADC2个12bitADC合计12了通道,外部通…...
Springboot项目中使用手机号短信验证码注册登录实现
文章目录 1. 功能概述2. 技术栈3. 实现步骤3.1 短信服务集成3.2 创建短信工具类3.3 验证码生成和存储3.4 控制器实现发送短信验证码手机号+验证码注册短信验证码登录3.5 服务层实现4. 前端实现4.1 API层4.2 手机号注册页面5. 最佳实践6. 总结本文将介绍如何在Spring Boot应用中…...
Python的GUI库选择指南(深度拓展)
前文我们分析了python的GUI库,有很多,面向应用场景也不尽相同,如何在使用过程中,选择合适的GUI库呢?可以查看:python有哪些常用的GUI(图形用户界面)库及选择指南-CSDN博客 初学者推…...
Kubernetes生命周期管理:深入理解 Pod 生命周期
云原生学习路线导航页(持续更新中) kubernetes学习系列快捷链接 Kubernetes架构原则和对象设计(一)Kubernetes架构原则和对象设计(二)Kubernetes架构原则和对象设计(三)Kubernetes控…...
JVM(12)——详解G1垃圾回收器
G1(Garbage-First)垃圾回收器。它是现代 Java 应用中默认的垃圾回收器(自 JDK 9 起),旨在提供一个高性能、可预测停顿时间(低延迟)的解决方案,尤其适合大内存(多GB甚至TB…...
Matplotlib vs Seaborn:选择与区别
相同点 都是Python数据可视化库:两者都用于创建统计图形和图表 基于Python生态系统:都与NumPy、Pandas等科学计算库良好集成 开源免费:两者都是开源项目,可自由使用 支持多种图表类型:都能创建折线图、柱状图、散点…...
TCP/UDP协议深度解析(一):UDP特性与TCP确认应答以及重传机制
🔍 开发者资源导航 🔍🏷️ 博客主页: 个人主页📚 专栏订阅: JavaEE全栈专栏 前言 在网络通信的世界里,传输层协议如同交通规则,决定了数据包如何从源头抵达目的地。其中UDP和TCP就…...