工具执行不仅仅依赖于输入参数和资源,还需要依赖于一些上下文信息,工具执行过程中所需的所有上下文信息都可以利用以参数注入的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是服务器与特定客户端之间的长连接句柄。

Contextlifespan_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_stateget_statedelete_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_componentsdisable_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())    
Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐