Claude Agent SDK
Integrate Claude Agent SDK with Acontext for session persistence
The ClaudeAgentStorage integration persists messages from the Claude Agent SDK to Acontext automatically. It stores only UserMessage and AssistantMessage in Anthropic format — SystemMessage, ResultMessage, and StreamEvent are used only for session-id resolution.
Quick Start
Install dependencies
pip install claude-agent-sdk acontext python-dotenvConfigure environment
ANTHROPIC_API_KEY=your_anthropic_key_here
ACONTEXT_API_KEY=sk-ac-your-api-keyRun agent with Acontext
import asyncio
from acontext import AcontextAsyncClient, ClaudeAgentStorage
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
async def main():
acontext_client = AcontextAsyncClient(api_key="sk-ac-your-api-key")
storage = ClaudeAgentStorage(client=acontext_client)
options = ClaudeAgentOptions(
extra_args={"replay-user-messages": None}, # include UserMessage in stream
)
async with ClaudeSDKClient(options=options) as claude_client:
await claude_client.query("What is the capital of France?")
async for message in claude_client.receive_response():
await storage.save_message(message)
asyncio.run(main())The replay-user-messages flag ensures UserMessage is included in the receive_response() stream so that both sides of the conversation are stored. Without it, only AssistantMessage appears in the stream and user messages won't be persisted.
Install dependencies
npm install @anthropic-ai/claude-agent-sdk @acontext/acontextConfigure environment
ANTHROPIC_API_KEY=your_anthropic_key_here
ACONTEXT_API_KEY=sk-ac-your-api-keyRun agent with Acontext
import { AcontextClient, ClaudeAgentStorage } from '@acontext/acontext';
import { query } from '@anthropic-ai/claude-agent-sdk';
const client = new AcontextClient({ apiKey: 'sk-ac-your-api-key' });
const storage = new ClaudeAgentStorage({ client });
const q = query({ prompt: 'What is the capital of France?',
options: {
extraArgs: { "replay-user-messages": null }, // include UserMessage in stream
}});
for await (const message of q) {
await storage.saveMessage(message);
}Session Handling
Auto-discovered session id
By default, ClaudeAgentStorage discovers the session id from the Claude Agent SDK.
When the Claude session id is a valid UUID, Acontext reuses it as the Acontext session id (via use_uuid). This means the Acontext session and the Claude session share the same id, making it easy to correlate them across systems.
If the Claude stream provides a non-UUID session id, it is ignored and Acontext generates its own session id instead.
storage = ClaudeAgentStorage(client=acontext_client)
# session_id is None until the init message arrives
async for message in claude_client.receive_response():
await storage.save_message(message)
# After init: storage.session_id matches the Claude session id
print(storage.session_id)const storage = new ClaudeAgentStorage({ client });
// sessionId is null until the init message arrives
for await (const message of q) {
await storage.saveMessage(message);
}
// After init: storage.sessionId matches the Claude session id
console.log(storage.sessionId);Explicit Acontext session
To correlate the Claude run with a specific Acontext session:
session = await acontext_client.sessions.create()
storage = ClaudeAgentStorage(client=acontext_client, session_id=session.id)
async for message in claude_client.receive_response():
await storage.save_message(message)const session = await client.sessions.create();
const storage = new ClaudeAgentStorage({ client, sessionId: session.id });
for await (const message of q) {
await storage.saveMessage(message);
}Options
Include thinking blocks
By default, ThinkingBlock content is omitted from stored messages. To include it as native thinking blocks:
storage = ClaudeAgentStorage(
client=acontext_client,
include_thinking=True,
)const storage = new ClaudeAgentStorage({
client,
includeThinking: true,
});Thinking blocks are stored as Anthropic thinking content blocks with their signature preserved. When retrieved in Anthropic or Gemini format, they round-trip as native thinking blocks. When retrieved in OpenAI format, thinking content is automatically downgraded to plain text.
The stored message meta will contain "has_thinking": True for observability.
Custom error handling
By default, API errors during storage are logged and swallowed so your message loop is never interrupted. To customize:
storage = ClaudeAgentStorage(
client=acontext_client,
on_error=lambda exc, msg: print(f"Storage failed: {exc}"),
)const storage = new ClaudeAgentStorage({
client,
onError: (err, blob) => console.error('Storage failed:', err),
});User identifier
Pass a user string to associate the Acontext session with a specific user. This value is forwarded to sessions.create() when the session is first created:
storage = ClaudeAgentStorage(
client=acontext_client,
user="alice@example.com",
)const storage = new ClaudeAgentStorage({
client,
user: 'alice@example.com',
});What gets stored
| Message type | Stored? | Notes |
|---|---|---|
| UserMessage | Yes | role: "user". String content wrapped as text block. ToolUseBlock skipped (invalid in user role). |
| AssistantMessage | Yes | role: "assistant". model stored in message meta. ToolResultBlock skipped (invalid in assistant role). If error field is set, its value is included in meta.error for observability. |
| SystemMessage | No | Used only for session-id resolution (subtype == "init" in Python, session_id field in TypeScript). |
| ResultMessage | No | Fallback session-id source. |
| StreamEvent | No | Fallback session-id source. |
Learn and Download Skills
Acontext can automatically learn skills from the stored messages — extracting patterns, user preferences, and domain knowledge into structured markdown files. You can then download these skills to your local filesystem:
# Create a learning space and learn from the session
space = await acontext_client.learning_spaces.create()
await acontext_client.learning_spaces.learn(space.id, session_id=storage.session_id)
# your agent runs as usual...
# ...
# Download learned skills to local
skills = await acontext_client.learning_spaces.list_skills(space.id)
for skill in skills:
result = await acontext_client.skills.download(skill_id=skill.id, path=f"./skills/{skill.name}")
print(f"Downloaded {result.name} to {result.dir_path}")
print(f"Files: {result.files}")// Create a learning space and learn from the session
const space = await client.learningSpaces.create();
await client.learningSpaces.learn({ spaceId: space.id, sessionId: storage.sessionId });
// your agent runs as usual...
// ...
// Download learned skills to local
const skills = await client.learningSpaces.listSkills(space.id);
for (const skill of skills) {
const result = await client.skills.download(skill.id, { path: `./skills/${skill.name}` });
console.log(`Downloaded ${result.name} to ${result.dirPath}`);
console.log(`Files: ${result.files}`);
}Next Steps
Last updated on