[FastMCP设计、原理与应用-16]Context:工具函数所需的所有上下文信息都在这里
工具执行不仅仅依赖于输入参数和资源,还需要依赖于一些上下文信息,工具执行过程中所需的所有上下文信息都可以利用以参数注入的fastmcp.server.Context来提供。不仅如此,这个Context还利用定义其中的方法实现了资源和提示词的消费、会话状态的维护以及资源可见性控制等操作。从服务端向客户端的反向请求和通知也是通过Context来完成的,对于FastMCP服务端编程来说,我个人认为这是最为重要的一个类型。
1. 上下文信息的提供
Context承载的针对当前执行上下的信息体现在如下所示的特性成员中。
@dataclass
class Context:
@property
def is_background_task(self) -> bool
@property
def task_id(self) -> str | None
@property
def origin_request_id(self) -> str | None
@property
def fastmcp(self) -> FastMCP
@property
def request_context(self) -> RequestContext[ServerSession, Any, Request] | None
@property
def lifespan_context(self) -> dict[str, Any]
@property
def transport(self) -> TransportType | None
@property
def client_id(self) -> str | None
@property
def request_id(self) -> str
@property
def session_id(self) -> str
@property
def session(self) -> ServerSession
Context提供了如上的特性,具体说明如下:
- is_background_task:当前是否以后台任务的形式在执行;
- task_id:如果
is_background_task返回True,返回当前任务的ID; - origin_request_id:如果
is_background_task返回True,返回触发当前后台任务的原始请求的ID,反之返回的就是当前请求的ID; - fastmcp:代表MCP服务器的
FastMCP对象; - request_context:当前请求上下文;
- lifespan_context:这是一个跨请求共享的字典。它的数据源头是我们定义的
lifespan函数; - transport:描述传输的
TransportType对象; - client_id: 客户端ID,侧重于客户端身份标识;
- request_id: 对应 JSON-RPC协议中的id。服务器必须拿着这个ID回复结果,客户端才能匹配上是哪个工具的返回;
- session/session_id: 代表当前的会话极其标识。Session是服务器与特定客户端之间的长连接句柄。
Context的lifespan_context特性的本质是 资源直通车。它将服务器启动时初始化的长期资源(如数据库连接、AI模型、配置对象,它们的生命周期于服务器绑定)安全地传递给每一个短期工具请求。它的源头是创建FastMCP时传入的lifespan函数。针对工具的每次调用,FastMCP都会创建一个新的Context对象,但是这个实例的lifespan_context特性是一个只读引用,下面的演示程序体现了这一点。
from typing import Self
from fastmcp import FastMCP
from fastmcp.server.lifespan import lifespan
from fastmcp.server import Context
from fastmcp.client import Client
import asyncio
log = []
class FakeDbConnection:
async def open(self)->Self:
log.append("DbConnection opened.")
return self
async def close(self):
log.append("DbConnection closed.")
lifespan_context = {}
@lifespan
async def app_lifespan(server):
db = await FakeDbConnection().open()
lifespan_context["db"] = db
try:
yield lifespan_context
finally:
await db.close()
server = FastMCP("Server", lifespan=app_lifespan)
@server.tool()
async def try_get_db(context: Context)-> bool:
assert context.lifespan_context is lifespan_context
db = context.lifespan_context.get("db")
return (db is not None) and isinstance(db, FakeDbConnection)
async def main():
async with Client(server) as client:
assert log == ["DbConnection opened."]
result = await client.call_tool("try_get_db")
assert bool(result.content[0].text) == True # type: ignore
assert log == ["DbConnection opened.", "DbConnection closed."]
asyncio.run(main())
2. 资源和提示词的消费
Context通常以参数注入的方式在工具函数中使用,对于工具来说,使用资源和提示词是常规的需求,所以Context定义了如下四个对应的方法分别完成资源和提示词列表的获取、提示词的渲染和资源的读取。
@dataclass
class Context:
async def list_resources(self) -> list[SDKResource]
async def list_prompts(self) -> list[SDKPrompt]
async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> GetPromptResult
async def read_resource(self, uri: str | AnyUrl) -> ResourceResult
3. 会话状态的维护
Session的一个核心的作用状态保持,为同一Session的多轮问答创建一个上下文,即上一轮问答设置的状态可以被下一轮问答读取到。会话状态的设置、读取和删除可以利用Context如下三个方法(set_state、get_state和delete_state)来完成。
@dataclass
class Context:
async def set_state(self, key: str, value: Any, *, serializable: bool = True) -> None:
async def get_state(self, key: str) -> Any
async def delete_state(self, key: str) -> None
如下的程序演示了统一会话中针对状态的读写和删除。
from fastmcp import FastMCP
from fastmcp.server import Context
from fastmcp.client import Client
import asyncio
server = FastMCP("Server")
@server.tool()
async def set_state(key: str, value: str, context: Context) -> None:
await context.set_state(key, value)
@server.tool()
async def get_state(key: str, context: Context) -> str | None:
return await context.get_state(key)
@server.tool()
async def delete_state(key: str, context: Context) -> None:
await context.delete_state(key)
async def main():
async with Client(server) as client:
await client.call_tool(
"set_state", arguments={"key": "color_preference", "value": "red"}
)
result = await client.call_tool(
"get_state", arguments={"key": "color_preference"}
)
assert result.content[0].text == "red" # type: ignore
await client.call_tool("delete_state", arguments={"key": "color_preference"})
result = await client.call_tool(
"get_state", arguments={"key": "color_preference"}
)
assert len(result.content) == 0
asyncio.run(main())
4. 资源可见性控制
Context如下所示的enable_components和disable_components方法可以在当前会话范围内控制组件的可见性,reset_visibility方法用于撤销针对组件可见性的设置。
@dataclass
class Context:
async def enable_components(
self,
*,
names: set[str] | None = None,
keys: set[str] | None = None,
version: VersionSpec | None = None,
tags: set[str] | None = None,
components: set[Literal["tool", "resource", "template", "prompt"]]
| None = None,
match_all: bool = False,
) -> None
async def disable_components(
self,
*,
names: set[str] | None = None,
keys: set[str] | None = None,
version: VersionSpec | None = None,
tags: set[str] | None = None,
components: set[Literal["tool", "resource", "template", "prompt"]]
| None = None,
match_all: bool = False,
) -> None
async def reset_visibility(self) -> None
如下的程序演示了利用通过调用工具set_tools_visibility对本Session范围内工具组件的可见性设置。
from fastmcp import FastMCP
from fastmcp.server import Context
from fastmcp.client import Client
import asyncio
server = FastMCP("Server")
@server.tool()
async def set_tools_visibility(
context: Context,
settings: dict[str, bool] | None = None,
reset: bool = False
) -> None:
if reset:
await context.reset_visibility()
return
settings = settings or {}
visible_tools = {
tool_name for tool_name, is_visible in settings.items() if is_visible
}
if len(visible_tools) > 0:
await context.enable_components(names=visible_tools,components={"tool"})
tools_to_invisible = set(settings.keys()) - visible_tools
if len(tools_to_invisible) > 0:
await context.disable_components(names=tools_to_invisible,components={"tool"})
@server.tool()
async def foo() -> None:
pass
@server.tool()
async def bar() -> None:
pass
@server.tool()
async def baz() -> None:
pass
async def main():
async with Client(server) as client:
await client.call_tool("set_tools_visibility", arguments={"settings": {"foo": False, "baz": False}})
tools = await client.list_tools()
tool_names = set(tool.name for tool in tools)
assert tool_names == {"bar", "set_tools_visibility"}
await client.call_tool("set_tools_visibility", arguments={"settings": {"foo": True}})
tools = await client.list_tools()
tool_names = set(tool.name for tool in tools)
assert tool_names == {"foo", "bar", "set_tools_visibility"}
await client.call_tool("set_tools_visibility", arguments={"reset": True})
tools = await client.list_tools()
tool_names = set(tool.name for tool in tools)
assert tool_names == {"foo", "bar", "baz", "set_tools_visibility"}
asyncio.run(main())
更多推荐
所有评论(0)