Acontext: Building Self-Learning AI Agents
Acontext is an open-source Context Data Platform for AI agents — think of it as Supabase, but purpose-built for agent context. It provides unified storage for messages, files, and skills, with built-in context engineering and observability so agents improve with every interaction.
Core Concepts
Acontext organizes agent context around five primitives:
- Session — Conversation thread that stores all messages, tool calls, and artifacts
- Disk — Filesystem-like artifact storage for agent-generated files
- Space — Knowledge base where distilled skills are indexed and retrieved
- Task — Auto-extracted execution unit with status (
pending → success / failed) - Skill Block — Reusable SOP derived from successful task completions
The learning loop is: Store → Observe → Learn → Act. Every session feeds experience into a Space, the background Experience Agent distills skills, and future runs start with proven patterns.
Project Structure
acontext-agent/
├── .env # API keys and config
├── requirements.txt # Python dependencies
├── main.py # Entry point
├── venv/ # Virtual environment (excluded from git)
├── agent/
│ ├── __init__.py
│ ├── client.py # Acontext client setup
│ ├── session.py # Session & message management
│ ├── disk.py # Artifact / file operations
│ └── skills.py # Space creation and skill search
├── tools/
│ ├── __init__.py
│ └── weather.py # Example tool implementation
└── README.mdSetup
Step 1: Create Project Directory
First, create the project directory structure:
# Create main project directory
mkdir acontext-agent
cd acontext-agent
# Create subdirectories
mkdir agent
mkdir tools
# Create __init__.py files to make them Python packages
touch agent/__init__.py
touch tools/__init__.pyStep 2: Create and Activate Virtual Environment
acontext SDK requires Python 3.11+ due to NotRequired type hints. Check with python --version or python3 --version.
Create a Python virtual environment to isolate your project dependencies:
# Create virtual environment
python -m venv venv
# Or use python3 if that's your command:
# python3 -m venv venv
# Activate virtual environment
# On macOS/Linux:
source venv/bin/activate
# On Windows:
# venv\Scripts\activateYou should see (venv) appear in your terminal prompt, indicating the virtual environment is active.
Step 3: Create requirements.txt
Create a requirements.txt file in your project root:
# requirements.txt
acontext>=0.1.0
openai>=1.0.0
python-dotenv>=1.0.0Step 4: Install Dependencies
Install all required packages in the virtual environment:
pip install -r requirements.txtStep 5: Get Your API Keys
Get a free Acontext API key at dash.acontext.io. You'll also need an OpenAI API key from platform.openai.com.
For self-hosted Docker backend (optional):
curl -fsSL https://install.acontext.io | sh && acontext server upStep 6: Configure Environment Variables
Create a .env file in your project root:
# .env
ACONTEXT_API_KEY=sk-ac-your-api-key-here
OPENAI_API_KEY=sk-your-openai-key
# For self-hosted Docker backend (optional)
# ACONTEXT_BASE_URL=http://localhost:8029/api/v1Step 7: Initialize the Acontext Client
Create agent/client.py to set up the Acontext client:
# agent/client.py
import os
from acontext import AcontextClient
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
def get_client() -> AcontextClient:
"""
Initialize and return an Acontext client.
Uses cloud API by default, or self-hosted URL if configured.
"""
api_key = os.getenv("ACONTEXT_API_KEY")
base_url = os.getenv("ACONTEXT_BASE_URL") # None uses cloud default
if base_url:
# Use self-hosted instance
return AcontextClient(base_url=base_url, api_key=api_key)
# Use cloud API
return AcontextClient(api_key=api_key)Step 8: Create Session Management Module
Create agent/session.py to manage conversation sessions and messages:
# agent/session.py
from acontext import AcontextClient
def create_session(client: AcontextClient, space_id: str | None = None) -> str:
"""
Create a new session to track a conversation.
Args:
client: Acontext client instance
space_id: Optional Space ID (reserved for future use)
Returns:
Session ID string
"""
# Note: space_id parameter not yet supported in SDK v0.1.x
session = client.sessions.create()
return session.id
def store_message(client: AcontextClient, session_id: str, role: str, content: str):
"""
Store a message in the session.
Messages are stored in OpenAI format and auto-converted on retrieval.
Args:
client: Acontext client instance
session_id: ID of the session
role: Message role ('user', 'assistant', 'system')
content: Message content
"""
client.sessions.store_message(
session_id=session_id,
blob={"role": role, "content": content},
format="openai",
)
def get_messages(client: AcontextClient, session_id: str, format: str = "openai") -> list:
"""
Retrieve all messages from a session.
Args:
client: Acontext client instance
session_id: ID of the session
format: Output format - 'openai', 'anthropic', or 'gemini'
Returns:
List of messages in requested format
"""
result = client.sessions.get_messages(session_id=session_id, format=format)
return result.items
def get_compressed_messages(client: AcontextClient, session_id: str) -> dict:
"""
Retrieve context-window-managed messages with automatic compression.
This applies strategies to keep messages within token limits.
Args:
client: Acontext client instance
session_id: ID of the session
Returns:
Dict with 'messages' list and 'tokens' count
"""
result = client.sessions.get_messages(
session_id=session_id,
edit_strategies=[
# Keep only the 3 most recent tool results
{"type": "remove_tool_result", "params": {"keep_recent_n_tool_results": 3}},
# Enforce 30k token limit
{"type": "token_limit", "params": {"limit_tokens": 30000}},
],
)
return {"messages": result.items, "tokens": result.this_time_tokens}
def get_session_summary(client: AcontextClient, session_id: str) -> str:
"""
Get a compact summary of the session for context injection.
Args:
client: Acontext client instance
session_id: ID of the session
Returns:
Summary text
"""
return client.sessions.get_session_summary(session_id=session_id, limit=5)Step 9: Create Disk Management Module
Create agent/disk.py for filesystem-like artifact storage:
# agent/disk.py
from acontext import AcontextClient
def create_disk(client: AcontextClient, name: str = "agent-artifacts") -> str:
"""
Create an isolated storage group for agent-generated files.
Think of this as creating a virtual filesystem for your agent.
Args:
client: Acontext client instance
name: Name for the disk storage (reserved for future use)
Returns:
Disk ID string
"""
# Note: name parameter not yet supported in SDK v0.1.x
disk = client.disks.create()
print(f"[Disk] Created disk: {disk.id}")
return disk.id
def upload_artifact(
client: AcontextClient, disk_id: str, filename: str, content: str
) -> str:
"""
Upload a text artifact (code, reports, markdown) to the agent's Disk.
Args:
client: Acontext client instance
disk_id: ID of the disk
filename: Name for the file
content: File content as string
Returns:
Artifact path string
"""
artifact = client.disks.artifacts.upsert(
disk_id=disk_id,
file=(filename, content.encode("utf-8"), "text/plain"),
file_path="/",
)
return f"{artifact.path}{artifact.filename}"
def list_artifacts(client: AcontextClient, disk_id: str) -> list[dict]:
"""
List all artifacts stored in the Disk.
Args:
client: Acontext client instance
disk_id: ID of the disk
Returns:
List of artifact metadata dicts
"""
result = client.disks.artifacts.list(disk_id=disk_id)
return [{"name": a.filename, "path": a.path} for a in result.artifacts]
def download_artifact(client: AcontextClient, disk_id: str, file_path: str, filename: str) -> str:
"""
Download artifact content as a string.
Args:
client: Acontext client instance
disk_id: ID of the disk
file_path: Path to the file within the disk
filename: Name of the file
Returns:
Artifact content as string
"""
content = client.disks.artifacts.download(
disk_id=disk_id,
file_path=file_path,
filename=filename,
)
return content.decode("utf-8")Step 10: Create Skill Space Module
Create agent/skills.py for learning from past experiences:
spaces API for skill learning is planned for a future SDK release. The functions below are stubbed to allow the agent to run. Once available, they will enable automatic skill distillation from successful task completions.
# agent/skills.py
# NOTE: The 'spaces' API is planned for a future SDK release.
# These functions are stubbed to allow the agent to run.
from acontext import AcontextClient
def create_space(client: AcontextClient, name: str = "agent-skills") -> str:
"""
Create a knowledge Space for storing and retrieving distilled skills.
This is where your agent's learned patterns accumulate over time.
Args:
client: Acontext client instance
name: Name for the skill space
Returns:
Placeholder space ID string
"""
# TODO: Implement when spaces API is available
# space = client.spaces.create(name=name)
# return space.id
print(f"[Skills] Space '{name}' (stubbed - spaces API coming soon)")
return "placeholder-space-id"
def search_skills(
client: AcontextClient, space_id: str, query: str, mode: str = "semantic"
) -> list[dict]:
"""
Search for relevant skills before executing a task.
This lets your agent recall past successful patterns.
Args:
client: Acontext client instance
space_id: ID of the skill space
query: Search query describing the current task
mode: Search mode ('semantic' or 'agentic')
Returns:
Empty list (stubbed until spaces API is available)
"""
# TODO: Implement when spaces API is available
# results = client.spaces.experience_search(
# space_id=space_id,
# query=query,
# mode=mode,
# )
# return [{"title": r.title, "content": r.content, "score": r.score} for r in results]
return []
def format_skills_as_context(skills: list[dict]) -> str:
"""
Format retrieved skills into a system-prompt-friendly string.
This prepares the context to inject into your agent's prompt.
Args:
skills: List of skill dicts from search_skills()
Returns:
Formatted string ready for prompt injection
"""
if not skills:
return ""
lines = ["## Relevant Past Experience\n"]
for skill in skills:
lines.append(f"### {skill['title']}\n{skill['content']}\n")
return "\n".join(lines)Step 11: Create Example Tool
Create tools/weather.py as an example tool for your agent:
# tools/weather.py
import json
# Tool definition in OpenAI function calling format
WEATHER_TOOL = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a city",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "City name"}
},
"required": ["location"],
},
},
}
def handle_get_weather(arguments: str) -> str:
"""
Handle weather tool calls from the agent.
Args:
arguments: JSON string with tool arguments
Returns:
JSON string with weather data
Note:
Replace this mock implementation with a real weather API call
(e.g., OpenWeatherMap, WeatherAPI, etc.)
"""
args = json.loads(arguments)
# Mock response - replace with actual API call
return json.dumps({
"location": args["location"],
"temp": "22°C",
"condition": "Sunny"
})Step 12: Create the Main Agent
Create main.py to tie everything together:
# main.py
import os
import json
from openai import OpenAI
from dotenv import load_dotenv
# Import our custom modules
from agent.client import get_client
from agent.session import (
create_session, store_message,
get_compressed_messages, get_session_summary,
)
from agent.disk import create_disk, upload_artifact
from agent.skills import create_space, search_skills, format_skills_as_context
from tools.weather import WEATHER_TOOL, handle_get_weather
# Load environment variables
load_dotenv()
# Initialize clients
ac = get_client() # Acontext client
openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) # OpenAI client
def run_agent(user_input: str, session_id: str, disk_id: str, space_id: str) -> str:
"""
Main agent execution flow with self-learning capabilities.
Args:
user_input: User's query or command
session_id: Current conversation session
disk_id: Storage disk for artifacts
space_id: Skill space for learning
Returns:
Agent's response string
"""
# STEP 1: Retrieve relevant skills from past successful experiences
# This allows the agent to learn from previous interactions
skills = search_skills(ac, space_id, user_input, mode="semantic")
skill_ctx = format_skills_as_context(skills)
# STEP 2: Build system prompt with injected skill context
# Past patterns guide current execution
system_prompt = (
"You are a helpful AI agent. Use tools when needed.\n" + skill_ctx
)
# STEP 3: Store the user message in session history
store_message(ac, session_id, "user", user_input)
# STEP 4: Retrieve context-managed history
# Acontext automatically handles token limits and compression
ctx = get_compressed_messages(ac, session_id)
print(f"[Context] Using {ctx['tokens']} tokens")
# STEP 5: Call LLM with full context
response = openai.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": system_prompt}] + ctx["messages"],
tools=[WEATHER_TOOL], # Available tools for the agent
)
msg = response.choices[0].message
# STEP 6: Handle tool calls if any
if msg.tool_calls:
# Store the assistant's message with tool_calls first
ac.sessions.store_message(
session_id=session_id,
blob=msg.model_dump(),
format="openai",
)
# Execute each tool and store results
for tc in msg.tool_calls:
result = handle_get_weather(tc.function.arguments)
# Store tool result with proper tool_call_id
ac.sessions.store_message(
session_id=session_id,
blob={
"role": "tool",
"tool_call_id": tc.id,
"content": result,
},
format="openai",
)
# Make a follow-up call with tool results
ctx2 = get_compressed_messages(ac, session_id)
response = openai.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": system_prompt}] + ctx2["messages"],
)
msg = response.choices[0].message
final_answer = msg.content or ""
# STEP 7: Store assistant reply and save as artifact
# This builds up the session history for future learning
store_message(ac, session_id, "assistant", final_answer)
upload_artifact(ac, disk_id, "last_response.txt", final_answer)
return final_answer
def main():
"""Initialize the agent and start interactive loop."""
# Initialize persistent resources
space_id = create_space(ac, "weather-agent-skills") # For learning
session_id = create_session(ac, space_id=space_id) # For conversation
disk_id = create_disk(ac, "weather-agent-artifacts") # For files
print("Agent ready. Type 'quit' to exit.\n")
# Interactive conversation loop
while True:
user_input = input("You: ").strip()
if user_input.lower() in ("quit", "exit"):
break
if not user_input:
continue
# Run the agent and display response
answer = run_agent(user_input, session_id, disk_id, space_id)
print(f"\nAgent: {answer}\n")
# Print session summary on exit
summary = get_session_summary(ac, session_id)
print(f"\n[Session Summary]\n{summary}")
if __name__ == "__main__":
main()Step 13: Run Your Agent
Now you're ready to run your self-learning agent!
Make sure your virtual environment is activated (you should see (venv) in your terminal prompt) before running the agent.
# Use the venv's Python directly to avoid shell alias issues
./venv/bin/python main.pyYou should see:
[Skills] Space 'weather-agent-skills' (stubbed - spaces API coming soon)
[Disk] Created disk: 86d2cc38-95f5-430c-a0d5-c6b0453dc945
Agent ready. Type 'quit' to exit.
You:You: hi
[Context] Using 2 tokens
Agent: Hello! How can I assist you today?
You: what's the weather in singapore
[Context] Using 17 tokens
Agent: The current weather in Singapore is 22°C and sunny.
You: what about tokyo?
[Context] Using 52 tokens
Agent: The current weather in Tokyo is 22°C and sunny.
You: which city did I ask about first?
[Context] Using 87 tokens
Agent: You first asked about the weather in Singapore.
You: quit
[Session Summary]
<task id="1" description="Get weather in Singapore">
<progress>
1. Retrieved weather data for Singapore
</progress>
</task>Notice how:
- Token count grows as conversation history accumulates (2 → 17 → 52 → 87 tokens)
- Context is preserved — the agent remembers Singapore was asked first
- Tool calls are handled — weather queries trigger the
get_weathertool - Session summary shows auto-extracted tasks on exit
Observability Dashboard
After running your agent, visit the Acontext dashboard to see what it learned:
- Cloud: dash.acontext.io
- Self-hosted:
http://localhost:8029
The dashboard shows:
- Auto-extracted tasks from conversations
- Success rates and performance metrics
- Distilled skills that were learned
- Step-by-step trace timelines for debugging
Skill distillation runs 10–30 seconds after a session ends. Refresh the dashboard to see newly learned SOPs appear in your Space.