### Install Development Dependencies Source: https://github.com/vfaddey/fastmcp-dishka/blob/main/README.md Use `uv sync --dev` to install all necessary dependencies for development, including testing and linting tools. ```bash uv sync --dev ``` -------------------------------- ### setup_dishka(container, app) Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Attaches the appropriate Dishka middleware to a FastMCP instance. Accepts either an AsyncContainer (installs DishkaAsyncMiddleware) or a synchronous Container (installs DishkaSyncMiddleware). Must be called once before the server starts. ```APIDOC ## setup_dishka(container, app) ### Description Attaches the appropriate Dishka middleware to a `FastMCP` instance. Accepts either an `AsyncContainer` (installs `DishkaAsyncMiddleware`) or a synchronous `Container` (installs `DishkaSyncMiddleware`). Must be called once before the server starts. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp_dishka import FastMCPProvider, setup_dishka class DatabaseService: async def fetch(self, key: str) -> str: return f"value_for_{key}" class AppProvider(Provider): db = provide(DatabaseService, scope=Scope.REQUEST) mcp = FastMCP("MyServer") container = make_async_container(AppProvider(), FastMCPProvider()) # Wire dependency injection into the FastMCP app setup_dishka(container=container, app=mcp) if __name__ == "__main__": mcp.run() ``` ### Response #### Success Response (200) None #### Response Example None ``` -------------------------------- ### Install Pre-commit Hooks Source: https://github.com/vfaddey/fastmcp-dishka/blob/main/README.md Set up pre-commit hooks to automatically check code quality before each commit by running `make pre-commit-install`. ```bash make pre-commit-install ``` -------------------------------- ### Setup Dishka Container and FastMCP Integration Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Initializes the FastMCP server with a Dishka container, including the `FastMCPProvider`. This setup should be done once at application startup. ```python mcp = FastMCP("NestedServer") container = make_async_container(AppProvider(), FastMCPProvider()) setup_dishka(container, mcp) ``` -------------------------------- ### Synchronous Container Support with FastMCP Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Use `make_container` instead of `make_async_container` when all providers and handlers are synchronous. DishkaSyncMiddleware is installed automatically. ```python from dishka import Provider, Scope, make_container, provide from fastmcp import FastMCP from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class CacheService: _store: dict[str, str] = {} def get(self, key: str) -> str | None: return self._store.get(key) def set(self, key: str, value: str) -> None: self._store[key] = value class AppProvider(Provider): cache = provide(CacheService, scope=Scope.APP) mcp = FastMCP("CacheServer") container = make_container(AppProvider(), FastMCPProvider()) # sync container setup_dishka(container, mcp) @mcp.tool @inject def cache_set(key: str, value: str, cache: FromDishka[CacheService]) -> str: cache.set(key, value) return f"stored {key}" @mcp.tool @inject def cache_get(key: str, cache: FromDishka[CacheService]) -> str: return cache.get(key) or "not found" ``` -------------------------------- ### Setup FastMCP with Dishka Source: https://github.com/vfaddey/fastmcp-dishka/blob/main/README.md Configure Dishka's container with `FastMCPProvider` and `setup_dishka` middleware to integrate with a FastMCP application. Ensure `@inject` is placed below FastMCP decorators. ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class GreetingService: def greet(self, name: str) -> str: return f"Hello, {name}!" class AppProvider(Provider): greeting = provide(GreetingService, scope=Scope.REQUEST) mcp = FastMCP("GreetMCP") container = make_async_container(AppProvider(), FastMCPProvider()) setup_dishka(container, mcp) @mcp.tool @inject async def greet(name: str, service: FromDishka[GreetingService]) -> str: return service.greet(name) ``` -------------------------------- ### Wire Dishka into a FastMCP app with setup_dishka Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Attaches Dishka middleware to a FastMCP instance. Must be called once before the server starts. Accepts either an `AsyncContainer` or a synchronous `Container`. ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp_dishka import FastMCPProvider, setup_dishka class DatabaseService: async def fetch(self, key: str) -> str: return f"value_for_{key}" class AppProvider(Provider): db = provide(DatabaseService, scope=Scope.REQUEST) mcp = FastMCP("MyServer") container = make_async_container(AppProvider(), FastMCPProvider()) # Wire dependency injection into the FastMCP app setup_dishka(container=container, app=mcp) if __name__ == "__main__": mcp.run() ``` -------------------------------- ### Run Linting and Formatting Source: https://github.com/vfaddey/fastmcp-dishka/blob/main/README.md Apply linting and code formatting rules to the project using `make lint` and `make format`. ```bash make lint ``` ```bash make format ``` -------------------------------- ### Run Tests with Coverage Source: https://github.com/vfaddey/fastmcp-dishka/blob/main/README.md Execute the test suite and generate a coverage report using the `make test` command. ```bash make test ``` -------------------------------- ### Injecting Dependencies into FastMCP Prompts Source: https://github.com/vfaddey/fastmcp-dishka/blob/main/README.md Use the `@inject` decorator with `FromDishka` to inject dependencies into FastMCP prompt functions. This allows prompts to access services managed by Dishka. ```python @mcp.prompt @inject async def welcome_prompt( name: str, service: FromDishka[GreetingService], ) -> str: return f"Write a short welcome message for: {service.greet(name)}" ``` -------------------------------- ### Resource and Prompt Injection with @inject Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt The `@inject` decorator works identically for resource templates and prompt handlers. Resources use URI template parameters, and prompts receive client call arguments. Both resolve `FromDishka` dependencies per request. ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class ContentRepo: async def get(self, item_id: str) -> str: return f"Content for item {item_id}" async def summarize(self, item_id: str) -> str: return f"Summary of item {item_id}" class AppProvider(Provider): repo = provide(ContentRepo, scope=Scope.REQUEST) mcp = FastMCP("ContentServer") container = make_async_container(AppProvider(), FastMCPProvider()) setup_dishka(container, mcp) # Resource: item_id comes from the URI template @mcp.resource("content://items/{item_id}") @inject async def item_resource( item_id: str, repo: FromDishka[ContentRepo], ) -> str: return await repo.get(item_id) # Prompt: name comes from the MCP client call arguments @mcp.prompt @inject async def item_summary_prompt( item_id: str, repo: FromDishka[ContentRepo], ) -> str: summary = await repo.summarize(item_id) return f"Please elaborate on this summary: {summary}" # Usage (programmatic): # await mcp.read_resource("content://items/42") → "Content for item 42" # await mcp.render_prompt("item_summary_prompt", {"item_id": "42"}) ``` -------------------------------- ### Implement Tool with Nested Resource Call Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt A tool that injects `Context` and `RequestIdService`. It demonstrates nested operation isolation, where the `rid.id` within the tool differs from that inside the nested resource call. ```python @mcp.tool @inject async def fetch_info( name: str, ctx: FromDishka[Context], rid: FromDishka[RequestIdService], ) -> str: # Nested resource read opens its OWN Scope.REQUEST container result = await ctx.read_resource(f"data://info/{name}") nested_content = result.contents[0].content # rid.id (tool scope) differs from the rid.id inside info_resource return f"tool_rid={rid.id} | resource={nested_content}" ``` -------------------------------- ### @inject Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt A decorator that wraps a tool, resource, or prompt handler and replaces every `FromDishka[T]`-annotated parameter with the resolved instance from the current request container. Automatically detects whether the function is async or sync and applies the correct injection strategy. Must be placed below the FastMCP decorator so FastMCP registers the original (unwrapped) signature. ```APIDOC ## @inject ### Description A decorator that wraps a tool, resource, or prompt handler and replaces every `FromDishka[T]`-annotated parameter with the resolved instance from the current request container. Automatically detects whether the function is async or sync and applies the correct injection strategy. **Must be placed below the FastMCP decorator** so FastMCP registers the original (unwrapped) signature. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class EmailService: async def send(self, to: str, body: str) -> bool: print(f"Sending '{body}' to {to}") return True class AppProvider(Provider): email = provide(EmailService, scope=Scope.REQUEST) mcp = FastMCP("EmailServer") container = make_async_container(AppProvider(), FastMCPProvider()) setup_dishka(container, mcp) @mcp.tool @inject # ← must be below @mcp.tool async def send_email( to: str, body: str, email_service: FromDishka[EmailService], # injected automatically ) -> str: success = await email_service.send(to, body) return "sent" if success else "failed" # list_tools() sees only {"to": str, "body": str} — FromDishka params are hidden ``` ### Response #### Success Response (200) None #### Response Example None ``` -------------------------------- ### FastMCPProvider for Request Scoped Objects Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Include FastMCPProvider to pre-register core FastMCP per-request objects. This allows handlers to inject Context, FastMCP, MiddlewareContext, and MCP parameter types without extra provider code. ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp.server.context import Context from fastmcp.server.middleware import MiddlewareContext from mcp import types as mt from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class AuditService: def record(self, method: str, tool: str) -> None: print(f"AUDIT: {method} called {tool}") class AppProvider(Provider): scope = Scope.APP @provide(scope=Scope.REQUEST) def audit(self, ctx: MiddlewareContext) -> AuditService: # MiddlewareContext is available because FastMCPProvider is included svc = AuditService() svc.record(ctx.method or "unknown", "init") return svc mcp = FastMCP("AuditServer") # FastMCPProvider must always be included in the container container = make_async_container(AppProvider(), FastMCPProvider()) setup_dishka(container, mcp) @mcp.tool @inject async def audited_tool( value: str, audit: FromDishka[AuditService], params: FromDishka[mt.CallToolRequestParams], # raw MCP params ) -> str: audit.record("call", params.name) return value.upper() ``` -------------------------------- ### Resolve FromDishka parameters with @inject decorator Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Wraps a handler to replace `FromDishka[T]` parameters with resolved instances. Must be placed below the FastMCP decorator (`@mcp.tool`). Automatically detects async or sync functions. ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class EmailService: async def send(self, to: str, body: str) -> bool: print(f"Sending '{body}' to {to}") return True class AppProvider(Provider): email = provide(EmailService, scope=Scope.REQUEST) mcp = FastMCP("EmailServer") container = make_async_container(AppProvider(), FastMCPProvider()) setup_dishka(container, mcp) @mcp.tool @inject # ← must be below @mcp.tool async def send_email( to: str, body: str, email_service: FromDishka[EmailService], # injected automatically ) -> str: success = await email_service.send(to, body) return "sent" if success else "failed" # list_tools() sees only {"to": str, "body": str} — FromDishka params are hidden ``` -------------------------------- ### Implement Resource with Dependency Injection Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt An MCP resource that injects a request-scoped `RequestIdService`. The `@inject` decorator must be placed below the FastMCP decorator. ```python @mcp.resource("data://info/{name}") @inject async def info_resource(name: str, rid: FromDishka[RequestIdService]) -> str: return f"{name}:{rid.id}" ``` -------------------------------- ### Define Request-Scoped Service and Providers Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Defines a service with request-specific instances and a Dishka provider for it. Ensure `Scope.REQUEST` is used for per-request isolation. ```python import asyncio from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp.server.context import Context from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class RequestIdService: def __init__(self) -> None: import uuid self.id = str(uuid.uuid4()) class AppProvider(Provider): req_id = provide(RequestIdService, scope=Scope.REQUEST) # new per request ``` -------------------------------- ### FromDishka[T] Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Re-exported directly from Dishka. Use it as a type annotation to mark function parameters that should be resolved from the Dishka container. The `@inject` decorator strips these from the public signature exposed to MCP clients. ```APIDOC ## FromDishka[T] ### Description Re-exported directly from Dishka. Use it as a type annotation to mark function parameters that should be resolved from the Dishka container. The `@inject` decorator strips these from the public signature exposed to MCP clients. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp.server.context import Context from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class RequestLogger: def log(self, msg: str) -> None: print(f"[LOG] {msg}") class AppProvider(Provider): logger = provide(RequestLogger, scope=Scope.REQUEST) mcp = FastMCP("LoggingServer") container = make_async_container(AppProvider(), FastMCPProvider()) setup_dishka(container, mcp) @mcp.tool @inject async def process( data: str, ctx: FromDishka[Context], # built-in FastMCP context via FastMCPProvider logger: FromDishka[RequestLogger], # custom service ) -> str: logger.log(f"Processing: {data}") return f"processed:{data}" ``` ### Response #### Success Response (200) None #### Response Example None ``` -------------------------------- ### Mark parameters for injection with FromDishka[T] Source: https://context7.com/vfaddey/fastmcp-dishka/llms.txt Use as a type annotation to mark parameters for resolution from the Dishka container. The `@inject` decorator hides these from the public signature exposed to MCP clients. ```python from dishka import Provider, Scope, make_async_container, provide from fastmcp import FastMCP from fastmcp.server.context import Context from fastmcp_dishka import FastMCPProvider, FromDishka, inject, setup_dishka class RequestLogger: def log(self, msg: str) -> None: print(f"[LOG] {msg}") class AppProvider(Provider): logger = provide(RequestLogger, scope=Scope.REQUEST) mcp = FastMCP("LoggingServer") container = make_async_container(AppProvider(), FastMCPProvider()) setup_dishka(container, mcp) @mcp.tool @inject async def process( data: str, ctx: FromDishka[Context], # built-in FastMCP context via FastMCPProvider logger: FromDishka[RequestLogger], # custom service ) -> str: logger.log(f"Processing: {data}") return f"processed:{data}" ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.