Context 对象允许你在 tool/resource/prompt 执行期间访问与请求同生命周期的一些实用功能——主要包括 logging、progress 和 request 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="Same as process(), but Context is injected")
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="Show auth context presence")
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}
Last modified on April 11, 2026