> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/bytedance/deer-flow/llms.txt
> Use this file to discover all available pages before exploring further.
# Thread Management
> Create and manage conversation threads with the LangGraph API
## Overview
Threads represent individual conversation sessions in DeerFlow. Each thread maintains its own:
* **Conversation history** - All messages exchanged
* **State data** - Title, artifacts, todos, uploads, viewed images
* **Isolated workspace** - Dedicated directories for files and outputs
* **Sandbox connection** - Persistent or ephemeral execution environment
## Creating Threads
Create a new thread using the LangGraph SDK:
```python theme={null}
from langgraph_sdk import get_client
client = get_client(url="http://localhost:2024")
# Create a thread
thread = await client.threads.create()
print(f"Thread ID: {thread['thread_id']}")
# Create with metadata
thread = await client.threads.create(
metadata={
"user_id": "user-123",
"project": "my-project",
"tags": ["research", "analysis"]
}
)
```
### Thread IDs
Thread IDs are UUIDs automatically generated by LangGraph. You can also provide your own:
```python theme={null}
import uuid
thread_id = str(uuid.uuid4())
thread = await client.threads.create(thread_id=thread_id)
```
Thread IDs must be unique. Creating a thread with an existing ID will fail.
## Listing Threads
Retrieve all threads with pagination:
```python theme={null}
# List all threads
threads = await client.threads.list()
# Paginated listing
threads = await client.threads.list(
limit=20,
offset=0
)
# Filter by metadata
threads = await client.threads.list(
metadata={"user_id": "user-123"}
)
```
## Getting Thread Details
Retrieve a specific thread:
```python theme={null}
thread = await client.threads.get(thread_id)
print(thread["thread_id"])
print(thread["metadata"])
print(thread["created_at"])
print(thread["updated_at"])
```
## Thread State
DeerFlow extends the standard LangGraph state with additional fields:
### State Schema
```python theme={null}
class ThreadState:
# Standard fields
messages: list[Message] # Conversation history
# DeerFlow extensions
sandbox: SandboxState # {sandbox_id: str}
thread_data: ThreadDataState # {workspace_path, uploads_path, outputs_path}
title: str | None # Auto-generated after first exchange
artifacts: list[str] # Files presented to user via present_files tool
todos: list | None # Task list (plan_mode only)
uploaded_files: list[dict] # User uploads with metadata
viewed_images: dict[str, ViewedImageData] # Image cache for vision models
```
### Getting State
```python theme={null}
# Get current state
state = await client.threads.get_state(thread_id)
print(state["values"]["messages"]) # All messages
print(state["values"]["title"]) # Thread title
print(state["values"]["artifacts"]) # Presented files
print(state["values"]["uploaded_files"]) # Uploads
```
### State Operations
Update thread state directly:
```python theme={null}
# Update state
await client.threads.update_state(
thread_id,
values={
"title": "Research Session",
"artifacts": ["analysis.pdf", "report.md"]
}
)
# Append to messages
await client.threads.update_state(
thread_id,
values={
"messages": [
{"role": "user", "content": "Additional context here"}
]
},
as_node="user" # Simulate user input
)
```
State updates trigger the middleware chain, so changes like adding uploads will be processed by `UploadsMiddleware`.
## Thread Configuration
Threads inherit runtime configuration from runs. You can specify different configs for each run:
```python theme={null}
# Run 1: Thinking mode enabled
await client.runs.create(
thread_id=thread_id,
assistant_id="lead_agent",
input={"messages": [{"role": "user", "content": "Analyze this"}]},
config={"configurable": {"thinking_enabled": True}}
)
# Run 2: Plan mode with different model
await client.runs.create(
thread_id=thread_id,
assistant_id="lead_agent",
input={"messages": [{"role": "user", "content": "Create a plan"}]},
config={
"configurable": {
"is_plan_mode": True,
"model_name": "claude-3-5-sonnet-20241022"
}
}
)
```
### Configuration Options
| Parameter | Type | Description |
| -------------------------- | ------ | ------------------------------------------------ |
| `thinking_enabled` | `bool` | Enable extended thinking (model must support it) |
| `reasoning_effort` | `str` | Reasoning level: "low", "medium", "high" |
| `model_name` | `str` | Override default model |
| `is_plan_mode` | `bool` | Enable task tracking with `write_todos` tool |
| `subagent_enabled` | `bool` | Enable `task` tool for sub-agent delegation |
| `max_concurrent_subagents` | `int` | Parallel sub-agent limit (default: 3) |
| `agent_name` | `str` | Use custom agent instead of lead\_agent |
## Thread Isolation
Each thread gets isolated directories managed by `ThreadDataMiddleware`:
```
backend/.deer-flow/threads/{thread_id}/
├── user-data/
│ ├── workspace/ # Working directory for file operations
│ ├── uploads/ # User-uploaded files
│ └── outputs/ # Files for download via present_files
```
### Virtual Path Mapping
Inside the sandbox, these paths are mapped:
| Physical Path | Virtual Path (in sandbox) |
| ---------------------------------------------- | --------------------------- |
| `.deer-flow/threads/{id}/user-data/workspace/` | `/mnt/user-data/workspace/` |
| `.deer-flow/threads/{id}/user-data/uploads/` | `/mnt/user-data/uploads/` |
| `.deer-flow/threads/{id}/user-data/outputs/` | `/mnt/user-data/outputs/` |
| `skills/` | `/mnt/skills/` |
The agent sees only virtual paths. Path translation happens automatically in sandbox tools.
## Working with Messages
### Message Format
DeerFlow uses LangChain message types:
```python theme={null}
# User message
{
"role": "user",
"content": "Hello!"
}
# AI message
{
"role": "assistant",
"content": "Hi there!"
}
# AI message with tool calls
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_abc123",
"type": "function",
"function": {
"name": "bash",
"arguments": '{"command": "ls -la"}'
}
}
]
}
# Tool result message
{
"role": "tool",
"tool_call_id": "call_abc123",
"content": "total 24\ndrwxr-xr-x..."
}
```
### Retrieving Messages
```python theme={null}
state = await client.threads.get_state(thread_id)
messages = state["values"]["messages"]
for msg in messages:
if msg["role"] == "user":
print(f"User: {msg['content']}")
elif msg["role"] == "assistant":
if msg.get("tool_calls"):
print(f"AI used {len(msg['tool_calls'])} tools")
else:
print(f"AI: {msg['content']}")
```
## Deleting Threads
Delete a thread and its associated data:
```python theme={null}
await client.threads.delete(thread_id)
```
This permanently deletes:
* All conversation history
* Thread state (title, artifacts, todos)
* Isolated workspace files
* Cannot be undone
## Interrupts
DeerFlow supports interrupts via the `ask_clarification` tool. When the agent needs user input:
```python theme={null}
# Agent calls ask_clarification tool
# ClarificationMiddleware intercepts and interrupts
# Check for interrupts
state = await client.threads.get_state(thread_id)
if state["next"] == ["__interrupt__"]:
print("Agent is waiting for input")
# Resume with user response
await client.threads.update_state(
thread_id,
values={
"messages": [
{"role": "user", "content": "Here's the clarification"}
]
}
)
```
### Interrupt Flow
1. Agent calls `ask_clarification` tool with a question
2. `ClarificationMiddleware` intercepts in `after_model` hook
3. Returns `Command(goto=END)` to halt execution
4. Thread state shows `next: ["__interrupt__"]`
5. Client provides clarification and resumes run
See [Streaming - Interrupt Events](/api/langgraph/streaming#interrupt-events) for handling interrupts in streaming mode.
## Checkpointing
LangGraph automatically checkpoints thread state after each step. This enables:
* **Time travel** - Rewind to any previous state
* **Branching** - Create alternate conversation paths
* **Recovery** - Resume after crashes
```python theme={null}
# Get state history
history = await client.threads.get_history(thread_id, limit=10)
for checkpoint in history:
print(f"Step {checkpoint['step']}: {checkpoint['values']['messages'][-1]}")
# Rewind to specific checkpoint
await client.threads.update_state(
thread_id,
values={},
checkpoint_id=checkpoint["checkpoint_id"]
)
```
## Example: Complete Thread Lifecycle
```python theme={null}
from langgraph_sdk import get_client
async def main():
client = get_client(url="http://localhost:2024")
# 1. Create thread
thread = await client.threads.create(
metadata={"project": "analysis"}
)
thread_id = thread["thread_id"]
# 2. Start conversation
await client.runs.create(
thread_id=thread_id,
assistant_id="lead_agent",
input={
"messages": [
{"role": "user", "content": "Analyze sales data in /mnt/user-data/uploads/sales.csv"}
]
},
config={
"configurable": {
"is_plan_mode": True,
"thinking_enabled": True
}
}
)
# 3. Get state
state = await client.threads.get_state(thread_id)
print(f"Title: {state['values']['title']}")
print(f"Artifacts: {state['values']['artifacts']}")
# 4. Continue conversation
await client.runs.create(
thread_id=thread_id,
assistant_id="lead_agent",
input={
"messages": [
{"role": "user", "content": "Now create a visualization"}
]
}
)
# 5. Cleanup
await client.threads.delete(thread_id)
```
## Best Practices
Store thread IDs in your database to maintain conversation continuity across sessions.
```python theme={null}
# Store in DB
thread_id = thread["thread_id"]
db.save_conversation(user_id, thread_id)
# Retrieve later
thread_id = db.get_conversation(user_id)
state = await client.threads.get_state(thread_id)
```
Use metadata to organize and filter threads:
```python theme={null}
await client.threads.create(
metadata={
"user_id": "user-123",
"type": "research",
"project": "q4-analysis",
"created_by": "web-app"
}
)
```
Implement a cleanup strategy for inactive threads:
```python theme={null}
# Delete threads older than 30 days
threads = await client.threads.list()
for thread in threads:
age = now - thread["updated_at"]
if age > timedelta(days=30):
await client.threads.delete(thread["thread_id"])
```
Always check for interrupt state before assuming completion:
```python theme={null}
state = await client.threads.get_state(thread_id)
if state["next"] == ["__interrupt__"]:
# Handle clarification request
last_message = state["values"]["messages"][-1]
if last_message.get("tool_calls"):
tool_call = last_message["tool_calls"][0]
if tool_call["function"]["name"] == "ask_clarification":
question = json.loads(tool_call["function"]["arguments"])["question"]
# Prompt user for clarification
```
## Next Steps
Learn how to stream agent responses in real-time
Upload files for agent processing
Enable task tracking with TodoList middleware
Understand how memory works across threads