跳转到主要内容
Context 对象允许你在 tool/resource/prompt 执行期间访问与请求同生命周期的一些实用功能——主要包括 loggingprogressrequest metadata 使用 get_context() 来获取当前活动的上下文。 注意get_context() 只能在一个活动的模型上下文协议 (MCP) 请求处理器内部使用。在其他位置调用会抛出 LookupError

获取上下文信息

from dedalus_mcp import get_context, tool

@tool(description="处理数据并记录日志")
async def process(data: str) -> str:
    ctx = get_context()
    await ctx.info("Processing", data={"bytes": len(data)})
    return "done"

自动注入(工具 + 依赖)

在工具(以及依赖中的可调用对象)中,标注为 Context 的参数可以由框架自动注入——无需手动调用 get_context()
from dedalus_mcp import Context, tool

@tool(description="与 process() 相同,但会注入 Context")
async def process_injected(data: str, ctx: Context) -> str:
    await ctx.info("Processing", data={"bytes": len(data)})
    return "done"

可用特性

特性API作用
日志记录ctx.debug(), ctx.info(), ctx.warning(), ctx.error(), ctx.log()向 Client 发送日志消息
请求元数据ctx.request_id, ctx.session_id, ctx.progress_token标识当前请求/会话以及进度令牌
服务器/运行时访问ctx.server, ctx.runtime访问运行时装配(如有)
认证上下文ctx.auth_context访问认证上下文(如果启用了授权)
进度ctx.report_progress(...), ctx.progress(...)发出进度通知(如果 Client 提供了进度令牌)
调度(可选)ctx.dispatch(...)通过已配置的调度后端发送带身份认证的 HTTP 请求(如果已配置)
Connection 解析(可选)ctx.resolve_client(...)通过已配置的解析器将 Connection 句柄解析为 Client(如果已配置)

请求元数据

from dedalus_mcp import get_context, tool

@tool(description="显示请求元数据")
async def my_tool() -> dict:
    ctx = get_context()
    return {
        "request_id": ctx.request_id,
        "session_id": ctx.session_id,          # 可能为 None(例如 stdio)
        "progress_token": ctx.progress_token,  # 如果客户端未请求进度,则可能为 None
    }

授权上下文

如果启用了授权,则会设置 ctx.auth_context;否则为 None
from dedalus_mcp import get_context, tool

@tool(description="显示授权上下文是否存在")
async def whoami() -> dict:
    ctx = get_context()
    if ctx.auth_context is None:
        return {"user": "anonymous"}
    return {"auth_context": "present"}

进度示例

import anyio
from dedalus_mcp import get_context, tool

@tool(description="处理文件并报告进度")
async def process_files(paths: list[str]) -> dict:
    ctx = get_context()
    processed = 0
    async with ctx.progress(total=len(paths)) as tracker:
        for _path in paths:
            await anyio.sleep(0.01)
            processed += 1
            await tracker.advance(1)

    return {"processed": processed}