diff --git a/docs/docs.json b/docs/docs.json
index f39a33aff7..de2d637fde 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -114,6 +114,7 @@
"icon": "stars",
"pages": [
"servers/tasks",
+ "servers/dependency-injection",
"servers/elicitation",
"servers/icons",
"servers/lifespan",
diff --git a/docs/servers/context.mdx b/docs/servers/context.mdx
index 3d6686cee6..78e71e53c7 100644
--- a/docs/servers/context.mdx
+++ b/docs/servers/context.mdx
@@ -10,7 +10,7 @@ import { VersionBadge } from '/snippets/version-badge.mdx'
When defining FastMCP [tools](/servers/tools), [resources](/servers/resources), resource templates, or [prompts](/servers/prompts), your functions might need to interact with the underlying MCP session or access advanced server capabilities. FastMCP provides the `Context` object for this purpose.
-FastMCP uses [Docket](https://github.com/chrisguidry/docket)'s dependency injection system for managing runtime dependencies. This page covers Context and the built-in dependencies; see [Custom Dependencies](#custom-dependencies) for creating your own.
+You access Context through FastMCP's dependency injection system. For other injectable values like HTTP requests, access tokens, and custom dependencies, see [Dependency Injection](/servers/dependency-injection).
## What Is Context?
@@ -388,7 +388,7 @@ async def session_info(ctx: Context) -> dict:
}
```
-For HTTP request access that works regardless of MCP session availability (when using HTTP transports), use the [HTTP request helpers](#http-requests) like `get_http_request()` and `get_http_headers()`.
+For HTTP request access that works regardless of MCP session availability (when using HTTP transports), use the [HTTP request helpers](/servers/dependency-injection#http-request) like `get_http_request()` and `get_http_headers()`.
#### Client Metadata
@@ -423,229 +423,3 @@ def send_email(to: str, subject: str, body: str, ctx: Context) -> str:
The MCP request is part of the low-level MCP SDK and intended for advanced use cases. Most users will not need to use it directly.
-## Runtime Dependencies
-
-### HTTP Requests
-
-
-
-The recommended way to access the current HTTP request is through the `get_http_request()` dependency function:
-
-```python {2, 3, 11}
-from fastmcp import FastMCP
-from fastmcp.server.dependencies import get_http_request
-from starlette.requests import Request
-
-mcp = FastMCP(name="HTTP Request Demo")
-
-@mcp.tool
-async def user_agent_info() -> dict:
- """Return information about the user agent."""
- # Get the HTTP request
- request: Request = get_http_request()
-
- # Access request data
- user_agent = request.headers.get("user-agent", "Unknown")
- client_ip = request.client.host if request.client else "Unknown"
-
- return {
- "user_agent": user_agent,
- "client_ip": client_ip,
- "path": request.url.path,
- }
-```
-
-This approach works anywhere within a request's execution flow, not just within your MCP function. It's useful when:
-
-1. You need access to HTTP information in helper functions
-2. You're calling nested functions that need HTTP request data
-3. You're working with middleware or other request processing code
-
-### HTTP Headers
-
-
-If you only need request headers and want to avoid potential errors, you can use the `get_http_headers()` helper:
-
-```python {2, 10}
-from fastmcp import FastMCP
-from fastmcp.server.dependencies import get_http_headers
-
-mcp = FastMCP(name="Headers Demo")
-
-@mcp.tool
-async def safe_header_info() -> dict:
- """Safely get header information without raising errors."""
- # Get headers (returns empty dict if no request context)
- headers = get_http_headers()
-
- # Get authorization header
- auth_header = headers.get("authorization", "")
- is_bearer = auth_header.startswith("Bearer ")
-
- return {
- "user_agent": headers.get("user-agent", "Unknown"),
- "content_type": headers.get("content-type", "Unknown"),
- "has_auth": bool(auth_header),
- "auth_type": "Bearer" if is_bearer else "Other" if auth_header else "None",
- "headers_count": len(headers)
- }
-```
-
-By default, `get_http_headers()` excludes problematic headers like `host` and `content-length`. To include all headers, use `get_http_headers(include_all=True)`.
-
-### Access Tokens
-
-
-
-When using authentication with your FastMCP server, you can access the authenticated user's access token information using the `get_access_token()` dependency function:
-
-```python {2, 10}
-from fastmcp import FastMCP
-from fastmcp.server.dependencies import get_access_token, AccessToken
-
-mcp = FastMCP(name="Auth Token Demo")
-
-@mcp.tool
-async def get_user_info() -> dict:
- """Get information about the authenticated user."""
- # Get the access token (None if not authenticated)
- token: AccessToken | None = get_access_token()
-
- if token is None:
- return {"authenticated": False}
-
- return {
- "authenticated": True,
- "client_id": token.client_id,
- "scopes": token.scopes,
- "expires_at": token.expires_at,
- "token_claims": token.claims, # JWT claims or custom token data
- }
-```
-
-This is particularly useful when you need to:
-
-1. **Access user identification** - Get the `client_id` or subject from token claims
-2. **Check permissions** - Verify scopes or custom claims before performing operations
-3. **Multi-tenant applications** - Extract tenant information from token claims
-4. **Audit logging** - Track which user performed which actions
-
-#### Working with Token Claims
-
-The `claims` field contains all the data from the original token (JWT claims for JWT tokens, or custom data for other token types):
-
-```python {2, 3, 9, 12, 15}
-from fastmcp import FastMCP
-from fastmcp.server.dependencies import get_access_token
-
-mcp = FastMCP(name="Multi-tenant Demo")
-
-@mcp.tool
-async def get_tenant_data(resource_id: str) -> dict:
- """Get tenant-specific data using token claims."""
- token: AccessToken | None = get_access_token()
-
- # Extract tenant ID from token claims
- tenant_id = token.claims.get("tenant_id") if token else None
-
- # Extract user ID from standard JWT subject claim
- user_id = token.claims.get("sub") if token else None
-
- # Use tenant and user info to authorize and filter data
- if not tenant_id:
- raise ValueError("No tenant information in token")
-
- return {
- "resource_id": resource_id,
- "tenant_id": tenant_id,
- "user_id": user_id,
- "data": f"Tenant-specific data for {tenant_id}",
- }
-```
-
-## Custom Dependencies
-
-
-
-FastMCP's dependency injection is powered by [Docket](https://github.com/chrisguidry/docket), which provides a flexible system for injecting values into your functions. Beyond the built-in dependencies like `CurrentContext()`, you can create your own.
-
-### Using `Depends()`
-
-The simplest way to create a custom dependency is with `Depends()`. Pass any callable (sync or async function, or async context manager) and its return value will be injected:
-
-```python
-from contextlib import asynccontextmanager
-from fastmcp import FastMCP
-from fastmcp.dependencies import Depends
-
-mcp = FastMCP(name="Custom Deps Demo")
-
-# Simple function dependency
-def get_config() -> dict:
- return {"api_url": "https://api.example.com", "timeout": 30}
-
-# Async function dependency
-async def get_user_id() -> int:
- return 42
-
-@mcp.tool
-async def fetch_data(
- query: str,
- config: dict = Depends(get_config),
- user_id: int = Depends(get_user_id),
-) -> str:
- return f"User {user_id} fetching '{query}' from {config['api_url']}"
-```
-
-Dependencies using `Depends()` are automatically excluded from the MCP schema—clients never see them as parameters.
-
-### Resource Management with Context Managers
-
-For dependencies that need cleanup (database connections, file handles, etc.), use an async context manager:
-
-```python
-from contextlib import asynccontextmanager
-from fastmcp import FastMCP
-from fastmcp.dependencies import Depends
-
-mcp = FastMCP(name="Resource Demo")
-
-@asynccontextmanager
-async def get_database():
- db = await connect_to_database()
- try:
- yield db
- finally:
- await db.close()
-
-@mcp.tool
-async def query_users(sql: str, db = Depends(get_database)) -> list:
- return await db.execute(sql)
-```
-
-The context manager's cleanup code runs after your function completes, even if an error occurs.
-
-### Nested Dependencies
-
-Dependencies can depend on other dependencies:
-
-```python
-from fastmcp import FastMCP
-from fastmcp.dependencies import Depends
-
-mcp = FastMCP(name="Nested Demo")
-
-def get_base_url() -> str:
- return "https://api.example.com"
-
-def get_api_client(base_url: str = Depends(get_base_url)) -> dict:
- return {"base_url": base_url, "version": "v1"}
-
-@mcp.tool
-async def call_api(endpoint: str, client: dict = Depends(get_api_client)) -> str:
- return f"Calling {client['base_url']}/{client['version']}/{endpoint}"
-```
-
-### Advanced: Subclassing `Dependency`
-
-For more complex dependency patterns—like dependencies that need access to Docket's execution context or require custom lifecycle management—you can subclass Docket's `Dependency` class. See the [Docket documentation on dependencies](https://chrisguidry.github.io/docket/dependencies/) for details.
diff --git a/docs/servers/dependency-injection.mdx b/docs/servers/dependency-injection.mdx
new file mode 100644
index 0000000000..d986bc52a2
--- /dev/null
+++ b/docs/servers/dependency-injection.mdx
@@ -0,0 +1,396 @@
+---
+title: Dependency Injection
+sidebarTitle: Dependencies
+description: Inject runtime values like HTTP requests, access tokens, and custom dependencies into your MCP components.
+icon: syringe
+tag: NEW
+---
+
+import { VersionBadge } from "/snippets/version-badge.mdx";
+
+FastMCP uses dependency injection to provide runtime values to your tools, resources, and prompts. Instead of passing context through every layer of your code, you declare what you need as parameter defaults—FastMCP resolves them automatically when your function runs.
+
+The dependency injection system is powered by [Docket](https://github.com/chrisguidry/docket). Core DI features like `Depends()` and `CurrentContext()` work without installing Docket. For background tasks and advanced task-related dependencies, install `fastmcp[tasks]`. For comprehensive coverage of dependency patterns, see the [Docket dependency documentation](https://chrisguidry.github.io/docket/dependencies/).
+
+
+Dependency parameters are automatically excluded from the MCP schema—clients never see them as callable parameters. This separation keeps your function signatures clean while giving you access to the runtime context you need.
+
+
+## How Dependency Injection Works
+
+Dependency injection in FastMCP follows a simple pattern: declare a parameter with a recognized type annotation or a dependency default value, and FastMCP injects the resolved value at runtime.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.context import Context
+
+mcp = FastMCP("Demo")
+
+
+@mcp.tool
+async def my_tool(query: str, ctx: Context) -> str:
+ await ctx.info(f"Processing: {query}")
+ return f"Results for: {query}"
+```
+
+When a client calls `my_tool`, they only see `query` as a parameter. The `ctx` parameter is injected automatically because it has a `Context` type annotation—FastMCP recognizes this and provides the active context for the request.
+
+This works identically for tools, resources, resource templates, and prompts.
+
+### Explicit Dependencies with CurrentContext
+
+For more explicit code, you can use `CurrentContext()` as a default value instead of relying on the type annotation:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import CurrentContext
+from fastmcp.server.context import Context
+
+mcp = FastMCP("Demo")
+
+
+@mcp.tool
+async def my_tool(query: str, ctx: Context = CurrentContext()) -> str:
+ await ctx.info(f"Processing: {query}")
+ return f"Results for: {query}"
+```
+
+Both approaches work identically. The type-annotation approach is more concise; the explicit `CurrentContext()` approach makes the dependency injection visible in the signature.
+
+## Built-in Dependencies
+
+### MCP Context
+
+The MCP Context provides logging, progress reporting, resource access, and other request-scoped operations. See [MCP Context](/servers/context) for the full API.
+
+**Dependency injection:** Use a `Context` type annotation (FastMCP injects automatically) or `CurrentContext()`:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.context import Context
+
+mcp = FastMCP("Demo")
+
+
+@mcp.tool
+async def process_data(data: str, ctx: Context) -> str:
+ await ctx.info(f"Processing: {data}")
+ return "Done"
+
+
+# Or explicitly with CurrentContext()
+from fastmcp.dependencies import CurrentContext
+
+@mcp.tool
+async def process_data(data: str, ctx: Context = CurrentContext()) -> str:
+ ...
+```
+
+**Function:** Use `get_context()` in helper functions or middleware:
+
+```python
+from fastmcp.server.dependencies import get_context
+
+async def log_something(message: str):
+ ctx = get_context()
+ await ctx.info(message)
+```
+
+### Server Instance
+
+
+
+Access the FastMCP server instance for introspection or server-level configuration.
+
+**Dependency injection:** Use `CurrentFastMCP()`:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import CurrentFastMCP
+
+mcp = FastMCP("Demo")
+
+
+@mcp.tool
+async def server_info(server: FastMCP = CurrentFastMCP()) -> str:
+ return f"Server: {server.name}"
+```
+
+**Function:** Use `get_server()`:
+
+```python
+from fastmcp.server.dependencies import get_server
+
+def get_server_name() -> str:
+ return get_server().name
+```
+
+### HTTP Request
+
+
+
+Access the Starlette Request when running over HTTP transports (SSE or Streamable HTTP).
+
+**Dependency injection:** Use `CurrentRequest()`:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import CurrentRequest
+from starlette.requests import Request
+
+mcp = FastMCP("Demo")
+
+
+@mcp.tool
+async def client_info(request: Request = CurrentRequest()) -> dict:
+ return {
+ "user_agent": request.headers.get("user-agent", "Unknown"),
+ "client_ip": request.client.host if request.client else "Unknown",
+ }
+```
+
+**Function:** Use `get_http_request()`:
+
+```python
+from fastmcp.server.dependencies import get_http_request
+
+def get_client_ip() -> str:
+ request = get_http_request()
+ return request.client.host if request.client else "Unknown"
+```
+
+
+Both raise `RuntimeError` when called outside an HTTP context (e.g., STDIO transport). Use HTTP Headers if you need graceful fallback.
+
+
+### HTTP Headers
+
+
+
+Access HTTP headers with graceful fallback—returns an empty dictionary when no HTTP request is available, making it safe for code that might run over any transport.
+
+**Dependency injection:** Use `CurrentHeaders()`:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import CurrentHeaders
+
+mcp = FastMCP("Demo")
+
+
+@mcp.tool
+async def get_auth_type(headers: dict = CurrentHeaders()) -> str:
+ auth = headers.get("authorization", "")
+ return "Bearer" if auth.startswith("Bearer ") else "None"
+```
+
+**Function:** Use `get_http_headers()`:
+
+```python
+from fastmcp.server.dependencies import get_http_headers
+
+def get_user_agent() -> str:
+ headers = get_http_headers()
+ return headers.get("user-agent", "Unknown")
+```
+
+By default, problematic headers like `host` and `content-length` are excluded. Use `get_http_headers(include_all=True)` to include all headers.
+
+### Access Token
+
+
+
+Access the authenticated user's token when your server uses authentication.
+
+**Dependency injection:** Use `CurrentAccessToken()` (raises if not authenticated):
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import CurrentAccessToken
+from fastmcp.server.auth import AccessToken
+
+mcp = FastMCP("Demo")
+
+
+@mcp.tool
+async def get_user_id(token: AccessToken = CurrentAccessToken()) -> str:
+ return token.claims.get("sub", "unknown")
+```
+
+**Function:** Use `get_access_token()` (returns `None` if not authenticated):
+
+```python
+from fastmcp.server.dependencies import get_access_token
+
+@mcp.tool
+async def get_user_info() -> dict:
+ token = get_access_token()
+ if token is None:
+ return {"authenticated": False}
+ return {"authenticated": True, "user": token.claims.get("sub")}
+```
+
+The `AccessToken` object provides:
+
+- **`client_id`**: The OAuth client identifier
+- **`scopes`**: List of granted permission scopes
+- **`expires_at`**: Token expiration timestamp (if available)
+- **`claims`**: Dictionary of all token claims (JWT claims or provider-specific data)
+
+### Background Task Dependencies
+
+
+
+For background task execution, FastMCP provides dependencies that integrate with [Docket](https://github.com/chrisguidry/docket). These require installing `fastmcp[tasks]`.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import CurrentDocket, CurrentWorker, Progress
+
+mcp = FastMCP("Task Demo")
+
+
+@mcp.tool(task=True)
+async def long_running_task(
+ data: str,
+ docket=CurrentDocket(),
+ worker=CurrentWorker(),
+ progress=Progress(),
+) -> str:
+ await progress.set_total(100)
+
+ for i in range(100):
+ # Process chunk...
+ await progress.increment()
+ await progress.set_message(f"Processing chunk {i + 1}")
+
+ return "Complete"
+```
+
+- **`CurrentDocket()`**: Access the Docket instance for scheduling additional background work
+- **`CurrentWorker()`**: Access the worker processing tasks (name, concurrency settings)
+- **`Progress()`**: Track task progress with atomic updates
+
+
+Task dependencies require `pip install 'fastmcp[tasks]'`. They're only available within task-enabled components (`task=True`). For comprehensive task patterns, see the [Docket documentation](https://chrisguidry.github.io/docket/dependencies/).
+
+
+## Custom Dependencies
+
+Beyond the built-in dependencies, you can create your own to inject configuration, database connections, API clients, or any other values your functions need.
+
+### Using Depends()
+
+The `Depends()` function wraps any callable and injects its return value. This works with synchronous functions, async functions, and async context managers.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import Depends
+
+mcp = FastMCP("Custom Deps Demo")
+
+
+def get_config() -> dict:
+ return {"api_url": "https://api.example.com", "timeout": 30}
+
+
+async def get_user_id() -> int:
+ # Could fetch from database, external service, etc.
+ return 42
+
+
+@mcp.tool
+async def fetch_data(
+ query: str,
+ config: dict = Depends(get_config),
+ user_id: int = Depends(get_user_id),
+) -> str:
+ return f"User {user_id} fetching '{query}' from {config['api_url']}"
+```
+
+### Caching
+
+Dependencies are cached per-request. If multiple parameters use the same dependency, or if nested dependencies share a common dependency, it's resolved once and the same instance is reused.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import Depends
+
+mcp = FastMCP("Caching Demo")
+
+
+def get_db_connection():
+ print("Connecting to database...") # Only printed once per request
+ return {"connection": "active"}
+
+
+def get_user_repo(db=Depends(get_db_connection)):
+ return {"db": db, "type": "user"}
+
+
+def get_order_repo(db=Depends(get_db_connection)):
+ return {"db": db, "type": "order"}
+
+
+@mcp.tool
+async def process_order(
+ order_id: str,
+ users=Depends(get_user_repo),
+ orders=Depends(get_order_repo),
+) -> str:
+ # Both repos share the same db connection
+ return f"Processed order {order_id}"
+```
+
+### Resource Management
+
+For dependencies that need cleanup—database connections, file handles, HTTP clients—use an async context manager. The cleanup code runs after your function completes, even if an error occurs.
+
+```python
+from contextlib import asynccontextmanager
+
+from fastmcp import FastMCP
+from fastmcp.dependencies import Depends
+
+mcp = FastMCP("Resource Demo")
+
+
+@asynccontextmanager
+async def get_database():
+ db = await connect_to_database()
+ try:
+ yield db
+ finally:
+ await db.close()
+
+
+@mcp.tool
+async def query_users(sql: str, db=Depends(get_database)) -> list:
+ return await db.execute(sql)
+```
+
+### Nested Dependencies
+
+Dependencies can depend on other dependencies. FastMCP resolves them in the correct order and applies caching across the dependency tree.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.dependencies import Depends
+
+mcp = FastMCP("Nested Demo")
+
+
+def get_base_url() -> str:
+ return "https://api.example.com"
+
+
+def get_api_client(base_url: str = Depends(get_base_url)) -> dict:
+ return {"base_url": base_url, "version": "v1"}
+
+
+@mcp.tool
+async def call_api(endpoint: str, client: dict = Depends(get_api_client)) -> str:
+ return f"Calling {client['base_url']}/{client['version']}/{endpoint}"
+```
+
+For advanced dependency patterns—like `TaskArgument()` for accessing task parameters, or custom `Dependency` subclasses—see the [Docket dependency documentation](https://chrisguidry.github.io/docket/dependencies/).
diff --git a/src/fastmcp/dependencies.py b/src/fastmcp/dependencies.py
index b9be7f1304..87d5367a92 100644
--- a/src/fastmcp/dependencies.py
+++ b/src/fastmcp/dependencies.py
@@ -17,18 +17,24 @@
from fastmcp.server.dependencies import (
+ CurrentAccessToken,
CurrentContext,
CurrentDocket,
CurrentFastMCP,
+ CurrentHeaders,
+ CurrentRequest,
CurrentWorker,
Progress,
ProgressLike,
)
__all__ = [
+ "CurrentAccessToken",
"CurrentContext",
"CurrentDocket",
"CurrentFastMCP",
+ "CurrentHeaders",
+ "CurrentRequest",
"CurrentWorker",
"Depends",
"Progress",
diff --git a/src/fastmcp/server/dependencies.py b/src/fastmcp/server/dependencies.py
index 0ecaac4aa5..1f20329180 100644
--- a/src/fastmcp/server/dependencies.py
+++ b/src/fastmcp/server/dependencies.py
@@ -42,9 +42,12 @@
__all__ = [
"AccessToken",
+ "CurrentAccessToken",
"CurrentContext",
"CurrentDocket",
"CurrentFastMCP",
+ "CurrentHeaders",
+ "CurrentRequest",
"CurrentWorker",
"Progress",
"get_access_token",
@@ -784,6 +787,116 @@ async def introspect(server: FastMCP = CurrentFastMCP()) -> str:
return cast(FastMCP, _CurrentFastMCP())
+class _CurrentRequest(Dependency): # type: ignore[misc]
+ """Async context manager for HTTP Request dependency."""
+
+ async def __aenter__(self) -> Request:
+ return get_http_request()
+
+ async def __aexit__(self, *args: object) -> None:
+ pass
+
+
+def CurrentRequest() -> Request:
+ """Get the current HTTP request.
+
+ This dependency provides access to the Starlette Request object for the
+ current HTTP request. Only available when running over HTTP transports
+ (SSE or Streamable HTTP).
+
+ Returns:
+ A dependency that resolves to the active Starlette Request
+
+ Raises:
+ RuntimeError: If no HTTP request in context (e.g., STDIO transport)
+
+ Example:
+ ```python
+ from fastmcp.server.dependencies import CurrentRequest
+ from starlette.requests import Request
+
+ @mcp.tool()
+ async def get_client_ip(request: Request = CurrentRequest()) -> str:
+ return request.client.host if request.client else "Unknown"
+ ```
+ """
+ return cast(Request, _CurrentRequest())
+
+
+class _CurrentHeaders(Dependency): # type: ignore[misc]
+ """Async context manager for HTTP Headers dependency."""
+
+ async def __aenter__(self) -> dict[str, str]:
+ return get_http_headers()
+
+ async def __aexit__(self, *args: object) -> None:
+ pass
+
+
+def CurrentHeaders() -> dict[str, str]:
+ """Get the current HTTP request headers.
+
+ This dependency provides access to the HTTP headers for the current request.
+ Returns an empty dictionary when no HTTP request is available, making it
+ safe to use in code that might run over any transport.
+
+ Returns:
+ A dependency that resolves to a dictionary of header name -> value
+
+ Example:
+ ```python
+ from fastmcp.server.dependencies import CurrentHeaders
+
+ @mcp.tool()
+ async def get_auth_type(headers: dict = CurrentHeaders()) -> str:
+ auth = headers.get("authorization", "")
+ return "Bearer" if auth.startswith("Bearer ") else "None"
+ ```
+ """
+ return cast(dict[str, str], _CurrentHeaders())
+
+
+class _CurrentAccessToken(Dependency): # type: ignore[misc]
+ """Async context manager for AccessToken dependency."""
+
+ async def __aenter__(self) -> AccessToken:
+ token = get_access_token()
+ if token is None:
+ raise RuntimeError(
+ "No access token found. Ensure authentication is configured "
+ "and the request is authenticated."
+ )
+ return token
+
+ async def __aexit__(self, *args: object) -> None:
+ pass
+
+
+def CurrentAccessToken() -> AccessToken:
+ """Get the current access token for the authenticated user.
+
+ This dependency provides access to the AccessToken for the current
+ authenticated request. Raises an error if no authentication is present.
+
+ Returns:
+ A dependency that resolves to the active AccessToken
+
+ Raises:
+ RuntimeError: If no authenticated user (use get_access_token() for optional)
+
+ Example:
+ ```python
+ from fastmcp.server.dependencies import CurrentAccessToken
+ from fastmcp.server.auth import AccessToken
+
+ @mcp.tool()
+ async def get_user_id(token: AccessToken = CurrentAccessToken()) -> str:
+ return token.claims.get("sub", "unknown")
+ ```
+ """
+ return cast(AccessToken, _CurrentAccessToken())
+
+
# --- Progress dependency ---