The Reliability Layer for Production AI Agents
Project description
Installation
pip install statebase
Requirements
- Python 3.8 or higher
requestslibrary (automatically installed)
Quick Start
from statebase import StateBase
# Initialize client
sb = StateBase(api_key="sb_live_xxxxxxxx")
# Create a session
session = sb.sessions.create(
agent_id="customer-support",
initial_state={"status": "new", "user_id": "user_123"}
)
# Update state
sb.sessions.update_state(
session_id=session.id,
state={"status": "in_progress"},
reasoning="User provided ticket details"
)
# Add a conversation turn
turn = sb.sessions.add_turn(
session_id=session.id,
input="My order hasn't arrived",
output="I'll help you track your order. Can you provide the order number?"
)
# Search memories
memories = sb.memory.search(
query="user communication preferences",
session_id=session.id
)
# Rollback if needed
sb.sessions.rollback(
session_id=session.id,
version=-1,
reason="Agent made an error"
)
Features
✅ Complete API Coverage
- Sessions (create, read, update, delete)
- State management with versioning
- Conversation turns
- Semantic memory (add, search)
- Decision traces
- Instant rollback
✅ Developer-Friendly
- Type hints for all methods
- Comprehensive error handling
- Automatic retries with exponential backoff
- Request/response logging (optional)
✅ Production-Ready
- Connection pooling
- Timeout configuration
- Rate limit handling
- Async support (optional)
Usage
Authentication
from statebase import StateBase
# Option 1: Pass API key directly
sb = StateBase(api_key="sb_live_xxxxxxxx")
# Option 2: Use environment variable
# Set STATEBASE_API_KEY in your environment
sb = StateBase()
# Option 3: Custom base URL (for self-hosted)
sb = StateBase(
api_key="your_key",
base_url="https://your-instance.com"
)
Sessions
# Create session
session = sb.sessions.create(
agent_id="support-bot",
initial_state={"user": "alice"},
ttl_seconds=86400 # 24 hours
)
# Get session
session = sb.sessions.get(session_id="sess_abc123")
# Update state
sb.sessions.update_state(
session_id="sess_abc123",
state={"step": "collect_info"},
reasoning="Moved to next step"
)
# Delete session
sb.sessions.delete(session_id="sess_abc123")
# List sessions
sessions = sb.sessions.list(
agent_id="support-bot",
limit=10
)
Conversation Turns
# Add turn
turn = sb.sessions.add_turn(
session_id="sess_abc123",
input="What's my order status?",
output="Your order #12345 is out for delivery",
metadata={"model": "gpt-4", "tokens": 150}
)
# Get turn history
turns = sb.sessions.get_turns(
session_id="sess_abc123",
limit=20
)
Memory
# Add memory
memory = sb.memory.add(
session_id="sess_abc123",
content="User prefers email over SMS",
type="preference",
tags=["communication", "preference"]
)
# Search memories
results = sb.memory.search(
query="how does user want to be contacted?",
session_id="sess_abc123",
top_k=5,
threshold=0.7
)
# Get memory by ID
memory = sb.memory.get(memory_id="mem_xyz789")
# Delete memory
sb.memory.delete(memory_id="mem_xyz789")
Rollback
# Rollback to previous version
sb.sessions.rollback(
session_id="sess_abc123",
version=-1, # or specific version number like 5
reason="Agent hallucinated"
)
# Rollback to specific version
sb.sessions.rollback(
session_id="sess_abc123",
version=3,
reason="Revert to known good state"
)
Traces
# List traces for a session
traces = sb.traces.list(
session_id="sess_abc123",
limit=100
)
# Get specific trace
trace = sb.traces.get(trace_id="trace_123")
# Filter traces by action
traces = sb.traces.list(
session_id="sess_abc123",
action="state_update"
)
Advanced Usage
Error Handling
from statebase import StateBase, StateBaseException
sb = StateBase(api_key="your_key")
try:
session = sb.sessions.create(agent_id="bot")
except StateBaseException as e:
print(f"Error: {e.message}")
print(f"Code: {e.error_code}")
print(f"Details: {e.details}")
Custom Configuration
sb = StateBase(
api_key="your_key",
base_url="https://api.statebase.org",
timeout=30, # seconds
max_retries=3,
retry_delay=1.0, # seconds
enable_logging=True
)
Async Support
from statebase import AsyncStateBase
async def main():
sb = AsyncStateBase(api_key="your_key")
session = await sb.sessions.create(
agent_id="async-bot"
)
await sb.sessions.update_state(
session_id=session.id,
state={"status": "processing"}
)
# Run with asyncio
import asyncio
asyncio.run(main())
Context Manager
with StateBase(api_key="your_key") as sb:
session = sb.sessions.create(agent_id="bot")
# Automatically handles cleanup
Integration Examples
With LangChain
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from statebase import StateBase
sb = StateBase(api_key="your_key")
llm = ChatOpenAI()
# Create session
session = sb.sessions.create(agent_id="langchain-bot")
# Your LangChain agent logic
def chat(message: str):
# Get context from StateBase
turns = sb.sessions.get_turns(session.id, limit=10)
context = "\n".join([f"User: {t.input}\nBot: {t.output}" for t in turns])
# Call LLM
response = llm.predict(f"{context}\nUser: {message}\nBot:")
# Save turn
sb.sessions.add_turn(
session_id=session.id,
input=message,
output=response
)
return response
With OpenAI
from openai import OpenAI
from statebase import StateBase
client = OpenAI()
sb = StateBase(api_key="your_key")
session = sb.sessions.create(agent_id="openai-bot")
def chat(message: str):
# Add user message
sb.sessions.add_turn(
session_id=session.id,
input=message,
output="" # Will update after LLM response
)
# Call OpenAI
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": message}]
)
output = response.choices[0].message.content
# Update with response
sb.sessions.update_state(
session_id=session.id,
state={"last_response": output}
)
return output
API Reference
Client
class StateBase:
def __init__(
self,
api_key: str = None,
base_url: str = "https://api.statebase.org",
timeout: int = 30,
max_retries: int = 3,
retry_delay: float = 1.0,
enable_logging: bool = False
)
Sessions
sb.sessions.create(agent_id: str, initial_state: dict = None, ttl_seconds: int = None) -> Session
sb.sessions.get(session_id: str) -> Session
sb.sessions.update_state(session_id: str, state: dict, reasoning: str = None) -> Session
sb.sessions.delete(session_id: str) -> None
sb.sessions.list(agent_id: str = None, limit: int = 10) -> List[Session]
sb.sessions.add_turn(session_id: str, input: str, output: str, metadata: dict = None) -> Turn
sb.sessions.get_turns(session_id: str, limit: int = 20) -> List[Turn]
sb.sessions.rollback(session_id: str, version: int, reason: str = None) -> Session
Memory
sb.memory.add(session_id: str, content: str, type: str = "general", tags: List[str] = None) -> Memory
sb.memory.search(query: str, session_id: str = None, top_k: int = 5, threshold: float = 0.7) -> List[Memory]
sb.memory.get(memory_id: str) -> Memory
sb.memory.delete(memory_id: str) -> None
Traces
sb.traces.list(session_id: str, limit: int = 100, action: str = None) -> List[Trace]
sb.traces.get(trace_id: str) -> Trace
Development
Setup
# Clone repository
git clone https://github.com/StateBase/statebase-python.git
cd statebase-python
# Create virtual environment
python -m venv venv
source venv/bin/activate
# Install dependencies
pip install -e ".[dev]"
Testing
# Run tests
pytest
# Run with coverage
pytest --cov=statebase
# Run specific test
pytest tests/test_sessions.py
Code Quality
# Format code
black statebase tests
# Sort imports
isort statebase tests
# Lint
flake8 statebase tests
# Type check
mypy statebase
Support
- Documentation: docs.statebase.org/python
- Issues: GitHub Issues
- Discord: discord.gg/statebase
- Email: support@statebase.org
License
MIT License - see LICENSE file for details.
Changelog
See CHANGELOG.md for version history.
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
statebase-0.4.1.tar.gz
(11.1 kB
view details)
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
statebase-0.4.1-py3-none-any.whl
(10.0 kB
view details)
File details
Details for the file statebase-0.4.1.tar.gz.
File metadata
- Download URL: statebase-0.4.1.tar.gz
- Upload date:
- Size: 11.1 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.10.11
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
961e0eb7f10d8ba0662eade30058d70e29adaffb52f27fc43aea05a3a64f2181
|
|
| MD5 |
375d641964ca23d3ecaa3f69e416d12d
|
|
| BLAKE2b-256 |
aa8b2befc6c814c66175f2f2c5d49ccc8755ead83599bb138551634749fcca31
|
File details
Details for the file statebase-0.4.1-py3-none-any.whl.
File metadata
- Download URL: statebase-0.4.1-py3-none-any.whl
- Upload date:
- Size: 10.0 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.10.11
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
63245d9e9983ef4bfeea488d0b4128543b4401a43662bdf75eef96a963b9abc2
|
|
| MD5 |
c787659082b86d0333e99d98f7c270e3
|
|
| BLAKE2b-256 |
bd64bd51dda1a1ec39225ce48c13386cdd45b5f45203bb84d105d995b22a1344
|