bagowix/dishka-fastmcp
GitHub: bagowix/dishka-fastmcp
dishka-fastmcp 将 dishka IoC 容器的依赖注入能力集成到 FastMCP 中，让 MCP 工具、资源和 prompt 支持带 scope 和资源清理的依赖解析。
Stars: 1 | Forks: 0
# dishka-fastmcp
[](https://github.com/bagowix/dishka-fastmcp/actions/workflows/ci.yml)
[](https://github.com/bagowix/dishka-fastmcp/tree/python-coverage-comment-action-data)
[](https://pypi.org/project/dishka-fastmcp/)
[](https://pypi.org/project/dishka-fastmcp/)
[](https://pypi.org/project/dishka-fastmcp/)
[](LICENSE)
[](docs/llms.txt)
[dishka](https://github.com/reagento/dishka) IoC 容器集成了
[FastMCP](https://github.com/jlowin/fastmcp)。在 MCP 工具、资源和 prompt 中将依赖声明为
`FromDishka[Service]`，并让 dishka 按请求解析它们 —— 具备真实的 scope、资源清理和模块化的 provider，与您的其余应用程序共享同一个容器。
```
from dishka import Provider, Scope, make_async_container, provide
from fastmcp import FastMCP
from dishka_fastmcp import FromDishka, dishka_lifespan, inject, setup_dishka
class Catalog:
_prices: dict[str, int] = {'book': 12, 'pen': 2}
def price(self, item: str) -> int:
return self._prices.get(item, 0)
class AppProvider(Provider):
catalog = provide(Catalog, scope=Scope.REQUEST)
container = make_async_container(AppProvider())
mcp = FastMCP('shop', lifespan=dishka_lifespan(container))
setup_dishka(container, mcp)
@mcp.tool
@inject
async def get_price(item: str, catalog: FromDishka[Catalog]) -> int:
return catalog.price(item)
if __name__ == '__main__':
mcp.run()
```
客户端看到的工具仅接收 `item` —— `catalog` 会在调用时被注入，并且永远不会出现在 schema 中。
## 安装
```
uv add dishka-fastmcp # or: pip install dishka-fastmcp
```
要求 Python 3.11+，`dishka>=1.10.1`，`fastmcp>=3.2.4,<4`。
## 工作原理
注册时间和执行时间是两个分离的考量点：
- **`@inject` 位于 FastMCP 装饰器下方。** `@mcp.tool` 根据函数签名构建 JSON
schema。`@inject` 会首先重写该签名，剔除所有的 `FromDishka` 参数，因此 LLM 看到的 schema 仅包含真正的面向客户端的参数。**顺序很重要** —— `@inject` 必须是内层装饰器。
- **`setup_dishka` 注册了一个 middleware**，它通过 `ContextVar` 暴露根容器和当前的 FastMCP 对象。然后，`@inject` 会在执行该 handler 的线程中，围绕它开启并完成 `Scope.REQUEST` 的清理。
这使得同步依赖的设置、使用和清理都在同一个 worker 线程中完成。
## Scope
| Scope | 边界 | 生命周期 |
|-------|----------|----------|
| `Scope.APP` | 整个服务器 | 由 `make_async_container` 创建；关闭时**由您**负责（见下文） |
| `Scope.REQUEST` | 一次工具调用 / 资源读取 / prompt 渲染 | 由 `@inject` 在 handler 周围开启并完成清理 |
**有意不支持** `Scope.SESSION`。FastMCP 的 middleware 暴露了会话启动钩子（`on_initialize`），但没有会话销毁钩子，因此具有会话生命周期的容器永远无法被确定性地销毁。与其提供一个在行为上静默等同于 REQUEST 的 SESSION scope，不如让 dishka-fastmcp 仅提供它能够完全支持的那两种 scope。如果 FastMCP 添加了会话销毁钩子，后续将会支持 SESSION。
### 关闭容器
`setup_dishka` 注册了 middleware，但并不拥有服务器的生命周期，因此它不会关闭根容器。上面示例中使用的 `dishka_lifespan(container)` 会在服务器停止时将其关闭（同步或异步），完成每个 `Scope.APP` provider 的清理。如果您已经有自己的 lifespan，请在其关闭路径中关闭容器（对于异步容器使用 `await container.close()`，对于同步容器使用 `container.close()`）。
FastMCP 可能会在不同的 worker 线程上执行同步 handler。因此，同步容器中的 `Scope.APP` 依赖必须是线程安全的，并且它们的清理操作不能依赖于创建它们的线程。请将诸如 `sqlite3.Connection` 等线程绑定的资源放在 `Scope.REQUEST` 中，在此 scope 下，dishka-fastmcp 会保证在同一个 worker 线程中进行创建、使用和销毁。
### 后台任务
**不支持** FastMCP 的 `task=True` 的 handler。工具调用会在工作排入队列后立即返回，因此当 worker 运行该 handler 时，请求 —— 以及随之而来的 REQUEST scope —— 已经结束了。在其中进行注入会引发 `DishkaFastMCPError` 错误。请保持 `FromDishka` 的 handler 绑定到请求级别；如果您需要后台处理，请在请求内部解析依赖，并将普通值传递给任务。
## 资源和 prompt
`@inject` 在资源和 prompt 上的工作方式相同 —— 依赖是按操作解析的：
```
@mcp.resource('users://{user_id}')
@inject
async def user(user_id: str, repo: FromDishka[UserRepo]) -> dict:
return await repo.get(user_id)
@mcp.prompt
@inject
async def summarize(text: str, summarizer: FromDishka[Summarizer]) -> str:
return await summarizer.run(text)
```
## 同步工具
FastMCP 在一个 worker 线程中运行同步工具（默认 `run_in_thread=True`）。
对于同步 handler，请使用**同步**容器（`make_container`）；请求容器会正确地传播到该 worker 线程中：
```
from dishka import make_container
container = make_container(AppProvider())
setup_dishka(container, mcp)
@mcp.tool
@inject
def compute(x: int, service: FromDishka[Calculator]) -> int:
return service.square(x)
```
异步 handler 需要异步容器（`make_async_container`）；混淆使用这两者会引发明确的错误。
## 访问 FastMCP 对象
添加 `FastMCPProvider`，即可通过 dishka 的 `from_context` 将当前请求的 FastMCP 对象暴露给您的依赖：
```
from fastmcp.server.context import Context
from dishka_fastmcp import FastMCPProvider
container = make_async_container(AppProvider(), FastMCPProvider())
@mcp.tool
@inject
async def notify(message: str, ctx: FromDishka[Context]) -> None:
await ctx.info(message)
```
`FastMCPProvider` 还暴露了 `FastMCP` 服务器以及正在进行的操作的原始请求参数
（`CallToolRequestParams`、`ReadResourceRequestParams`、`GetPromptRequestParams`）。
## 对比
### 与 `fastmcp.dependencies.Depends` 的对比
FastMCP 自带的 `Depends` 会注入一个按调用生成的值，并将其对 schema 隐藏 —— 这对于简单的场景已经足够。dishka 提供了 `Depends` 所不具备的功能：**带资源清理的 scope**、**模块化的 provider**，以及**与应用程序其余部分共享的一个容器**（为您的 FastAPI 或 FastStream 代码提供数据的同一个图谱，也可以为您的 MCP 服务器提供数据）。当您的 MCP 服务器是更大的 dishka 应用程序的一部分时，或者当您的依赖项拥有必须在每次请求时进行创建和销毁的资源时，请选择 dishka-fastmcp。
### 与 [`fastmcp-dishka`](https://github.com/vfaddey/fastmcp-dishka) 的对比
`fastmcp-dishka`（Apache-2.0）是涵盖相同领域的先前作品，这两个包都基于 dishka 公开的 `wrap_injection` 辅助函数构建。不同之处在于：
- **我们能够支持的 Scope。** 出于上述原因，dishka-fastmcp 仅提供 `APP` 和 `REQUEST`。而 `fastmcp-dishka` 还暴露了 `SESSION`，但在当前的 FastMCP 中，它是按请求而不是按会话解析的。
- **不与 FastMCP 内部耦合。** 请求容器通过本包拥有的 `ContextVar` 进行传递，而不是通过私有的 FastMCP 属性。
- **质量标准。** 100% 的测试覆盖率，在严格模式下的 `mypy` 和 `pyright`，以及跨越 Python 3.11–3.14 的 CI 矩阵。
## 许可证
MIT —— 见 [LICENSE](LICENSE)。
标签：FastMCP, MCP, Python, SOC Prime, 依赖注入, 开发工具, 控制反转, 无后门, 逆向工具