python-sdk/docs/experimental/tasks.md

189 lines
6.7 KiB
Markdown

# Tasks
!!! warning "Deprecated"
The experimental tasks API is deprecated and will be removed in mcp 2.0.
Tasks (SEP-1686) were removed from the MCP specification and are expected
to return as a separate MCP extension in a future release.
Tasks enable asynchronous request handling in MCP. Instead of blocking until an operation completes, the receiver creates a task, returns immediately, and the requestor polls for the result.
## When to Use Tasks
Tasks are designed for operations that:
- Take significant time (seconds to minutes)
- Need progress updates during execution
- Require user input mid-execution (elicitation, sampling)
- Should run without blocking the requestor
Common use cases:
- Long-running data processing
- Multi-step workflows with user confirmation
- LLM-powered operations requiring sampling
- OAuth flows requiring user browser interaction
## Task Lifecycle
```text
┌─────────────┐
│ working │
└──────┬──────┘
│
┌────────────┼────────────┐
│ │ │
▼ ▼ ▼
┌────────────┐ ┌───────────┐ ┌───────────┐
│ completed │ │ failed │ │ cancelled │
└────────────┘ └───────────┘ └───────────┘
▲
│
┌────────┴────────┐
│ input_required │◄──────┐
└────────┬────────┘ │
│ │
└────────────────┘
```
| Status | Description |
|--------|-------------|
| `working` | Task is being processed |
| `input_required` | Receiver needs input from requestor (elicitation/sampling) |
| `completed` | Task finished successfully |
| `failed` | Task encountered an error |
| `cancelled` | Task was cancelled by requestor |
Terminal states (`completed`, `failed`, `cancelled`) are final—tasks cannot transition out of them.
## Bidirectional Flow
Tasks work in both directions:
**Client → Server** (most common):
```text
Client Server
│ │
│── tools/call (task) ──────────────>│ Creates task
│<── CreateTaskResult ───────────────│
│ │
│── tasks/get ──────────────────────>│
│<── status: working ────────────────│
│ │ ... work continues ...
│── tasks/get ──────────────────────>│
│<── status: completed ──────────────│
│ │
│── tasks/result ───────────────────>│
│<── CallToolResult ─────────────────│
```
**Server → Client** (for elicitation/sampling):
```text
Server Client
│ │
│── elicitation/create (task) ──────>│ Creates task
│<── CreateTaskResult ───────────────│
│ │
│── tasks/get ──────────────────────>│
│<── status: working ────────────────│
│ │ ... user interaction ...
│── tasks/get ──────────────────────>│
│<── status: completed ──────────────│
│ │
│── tasks/result ───────────────────>│
│<── ElicitResult ───────────────────│
```
## Key Concepts
### Task Metadata
When augmenting a request with task execution, include `TaskMetadata`:
```python
from mcp.types import TaskMetadata
task = TaskMetadata(ttl=60000) # TTL in milliseconds
```
The `ttl` (time-to-live) specifies how long the task and result are retained after completion.
### Task Store
Servers persist task state in a `TaskStore`. The SDK provides `InMemoryTaskStore` for development:
```python
from mcp.shared.experimental.tasks import InMemoryTaskStore
store = InMemoryTaskStore()
```
For production, implement `TaskStore` with a database or distributed cache.
### Capabilities
Both servers and clients declare task support through capabilities:
**Server capabilities:**
- `tasks.requests.tools.call` - Server accepts task-augmented tool calls
**Client capabilities:**
- `tasks.requests.sampling.createMessage` - Client accepts task-augmented sampling
- `tasks.requests.elicitation.create` - Client accepts task-augmented elicitation
The SDK manages these automatically when you enable task support.
## Quick Example
**Server** (simplified API):
```python
from mcp.server import Server
from mcp.server.experimental.task_context import ServerTaskContext
from mcp.types import CallToolResult, TextContent, TASK_REQUIRED
server = Server("my-server")
server.experimental.enable_tasks() # One-line setup
@server.call_tool()
async def handle_tool(name: str, arguments: dict):
ctx = server.request_context
ctx.experimental.validate_task_mode(TASK_REQUIRED)
async def work(task: ServerTaskContext):
await task.update_status("Processing...")
# ... do work ...
return CallToolResult(content=[TextContent(type="text", text="Done!")])
return await ctx.experimental.run_task(work)
```
**Client:**
```python
from mcp.client.session import ClientSession
from mcp.types import CallToolResult
async with ClientSession(read, write) as session:
await session.initialize()
# Call tool as task
result = await session.experimental.call_tool_as_task("my_tool", {"arg": "value"})
task_id = result.task.taskId
# Poll until done
async for status in session.experimental.poll_task(task_id):
print(f"Status: {status.status}")
# Get result
final = await session.experimental.get_task_result(task_id, CallToolResult)
```
## Next Steps
- [Server Implementation](tasks-server.md) - Build task-supporting servers
- [Client Usage](tasks-client.md) - Call and poll tasks from clients