第四章 FastMCP 服务器框架之提示词注册机制
FastMCP框架通过@prompt装饰器实现提示词的注册与管理。核心流程包括:1) @prompt装饰器接收元数据参数并包装目标函数;2) Prompt.from_function将函数转换为Prompt对象,自动提取参数元数据;3) 通过find_context_parameter实现上下文注入;4) 注册时检查重复提示词并发出警告。框架支持同步和异步函数,利用Pydantic进行参数验证,并
目录
提示词注册流程概述
FastMCP框架中的提示词注册机制通过@prompt装饰器将普通Python函数转换为可管理的Prompt对象。该流程始于FastMCP服务器实例的prompt方法,该方法作为装饰器工厂,接收元数据参数(如名称、标题、描述、图标)并返回一个装饰器函数。当此装饰器应用于目标函数时,它会调用Prompt.from_function类方法,将函数及其元数据转换为Prompt实例,最后通过PromptManager的add_prompt方法将该实例注册到管理器中,完成整个注册过程。
本节来源
@prompt装饰器与Prompt.from_function方法
@prompt装饰器是提示词注册的核心入口。它首先验证装饰器是否被正确调用(即@prompt()而非@prompt),然后定义一个内部装饰器函数。该内部函数接收目标函数作为参数,调用Prompt.from_function方法,并将生成的Prompt对象传递给FastMCP服务器的add_prompt方法。
Prompt.from_function是Prompt类的类方法,负责将普通函数转换为Prompt对象。其核心步骤包括:确定提示词名称(优先使用name参数,否则使用函数名)、通过find_context_parameter查找上下文参数、利用func_metadata工具提取函数参数的元数据、将参数元数据转换为PromptArgument列表,并最终使用validate_call包装原函数以确保参数验证,然后构造并返回Prompt实例。
图源
本节来源
函数签名解析与参数元数据提取
func_metadata函数是参数元数据提取的关键。它接收一个函数和一个可选的skip_names列表(用于排除上下文参数等),并返回一个FuncMetadata对象。该函数首先获取函数的类型化签名,然后遍历所有参数。对于每个参数,它会检查其注解(annotation),处理特殊情况(如None或未注解的参数),并创建一个FieldInfo对象。这些FieldInfo对象随后被用来动态创建一个名为{func.__name__}Arguments的Pydantic模型。
Prompt.from_function方法利用func_metadata返回的模型,通过调用model_json_schema()方法生成JSON Schema。然后,它遍历Schema中的properties,为每个参数创建一个PromptArgument对象,其中包含参数名、描述和是否必需等信息。这个PromptArgument列表最终被赋值给Prompt对象的arguments字段,使得提示词的参数信息可以被外部系统查询和使用。
图源
本节来源
上下文注入机制
上下文注入机制允许提示词函数访问服务器的上下文信息,如会话、日志记录和进度报告功能。find_context_parameter函数负责在函数签名中查找类型为Context或其子类的参数。它通过typing.get_type_hints获取函数的类型提示,然后检查每个参数的注解。它能处理直接类型(ctx: Context)和泛型类型(ctx: Optional[Context])。
一旦find_context_parameter返回了上下文参数名(如ctx),Prompt.from_function会将此名称存储在Prompt对象的context_kwarg字段中。当Prompt的render方法被调用时,它会使用inject_context函数。该函数检查context_kwarg是否不为None且传入的context对象有效,如果条件满足,则将context对象作为关键字参数注入到调用原函数的参数字典中,从而实现上下文的自动注入。
图源
本节来源
重复提示词检测与警告
PromptManager类负责管理所有已注册的提示词,其add_prompt方法内置了重复检测机制。当一个新的Prompt对象被添加时,add_prompt会首先检查_prompts字典中是否已存在同名的提示词。如果存在,它会根据warn_on_duplicate_prompts标志位决定是否记录警告日志。无论是否记录警告,该方法都会直接返回已存在的Prompt对象,从而避免了重复注册和潜在的内存泄漏。
此机制确保了提示词名称的唯一性,防止了因意外重复注册而导致的逻辑错误。用户可以通过在创建PromptManager实例时将warn_on_duplicate_prompts设置为False来禁用警告,但提示词的去重逻辑依然会执行。
本节来源
同步与异步提示词示例
以下代码示例展示了如何使用@prompt装饰器定义同步和异步提示词函数,并通过icons、title、description等参数进行元数据配置。
# 定义一个同步提示词
@server.prompt(
title="分析表格",
description="根据表名生成分析该表结构的提示词",
icons=[Icon(url="https://example.com/chart-icon.png")]
)
def analyze_table(table_name: str) -> list[Message]:
schema = read_table_schema(table_name)
return [
UserMessage(content=f"分析此表结构:\n{schema}")
]
# 定义一个异步提示词
@server.prompt(
title="分析文件",
description="读取文件内容并生成分析该文件的提示词",
icons=[Icon(url="https://example.com/file-icon.png")]
)
async def analyze_file(path: str) -> list[Message]:
content = await read_file(path)
return [
UserMessage(content=TextContent(type="text", text=content))
]
# 定义一个需要上下文的提示词
@server.prompt(title="记录日志")
async def log_operation(operation: str, ctx: Context) -> list[Message]:
await ctx.info(f"操作 {operation} 开始")
# ... 执行操作 ...
await ctx.info(f"操作 {operation} 完成")
return [AssistantMessage(content="操作已记录")]
本节来源
火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。
更多推荐
所有评论(0)