vouchstone-sdk 1.2.0

Python SDK for the Vouchstone Enterprise AI Agent Platform — build, deploy, and govern production AI agents

Vouchstone SDK

Python SDK for the Vouchstone Enterprise AI Agent Platform — build, deploy, and govern production AI agents with persistent memory, full auditability, and enterprise-grade controls.

Vouchstone LLC | Website | Docs | GitHub

---

What Is Vouchstone?

Vouchstone is the first Accountable AI Engineering Platform — a control plane + data plane architecture for enterprises that need AI agents they can trust, audit, and govern. The platform provides:

• AI Agent Lifecycle — Create, deploy, monitor, and retire agents through a managed control plane
• 5-Layer Persistent Memory — Working, Episodic, Semantic, Procedural, and Meta-Memory
• Enterprise Governance — ABAC policies, RACI matrices, cost governance, shadow mode, compliance packs
• Multi-Tenant SaaS — Tenant-isolated data, Stripe billing, SAML/OIDC federation
• 70+ Integrations — Slack, Teams, JIRA, GitHub, Salesforce, Snowflake, and more

This SDK lets you build custom agents that run on the Vouchstone data plane and interact with the control plane APIs.

---

Install

pip install vouchstone-sdk

Optional Extras

# Working memory (Redis-backed per-session context)
pip install vouchstone-sdk[redis]

# Semantic memory (ChromaDB vector search)
pip install vouchstone-sdk[vector]

# Procedural memory (Neo4j skill graph)
pip install vouchstone-sdk[graph]

# Everything
pip install vouchstone-sdk[all]

Requirements

• Python 3.10+
• A Vouchstone control plane instance (self-hosted or cloud)
• API key from your Vouchstone tenant

---

Quick Start

1. Build a Custom Agent

from vouchstone_sdk import Agent, AgentConfig, Message, AgentResponse, MemoryContext

class DataMigrationAgent(Agent):
    async def run(self, message: Message, context: MemoryContext) -> AgentResponse:
        # context.working_memory   — current session turns
        # context.episodic_context — past session traces
        # context.semantic_entities — known entities (tech, people, systems)
        # context.procedural_skills — learned procedures
        # context.scratchpad       — per-session key-value store

        # Your LLM call here
        response = await self.llm.complete(
            system=f"You are {self.config.name}.",
            messages=[{"role": "user", "content": message.content}],
        )
        return AgentResponse(content=response)

config = AgentConfig(
    name="Data Migration Agent",
    model="claude-sonnet-4-20250514",
    system_prompt="You help enterprises migrate data between systems.",
)

agent = DataMigrationAgent(config)
await agent.initialize(
    agent_id="agent-123",
    redis_url="redis://localhost:6379",
    vector_db_url="http://localhost:8002",
    graph_db_url="bolt://localhost:7687",
)

session = agent.start_session()
response = await agent.process(Message(content="Migrate PostgreSQL to Snowflake"))
print(response.content)

await agent.end_session()
await agent.close()

2. Connect to the Control Plane

from vouchstone_sdk import VouchstoneClient

async with VouchstoneClient(
    api_key="your-api-key",
    control_plane_url="https://api.vouchstone.ai",
    tenant_id="your-tenant-id",
) as client:
    # List all agents in your tenant
    agents = await client.list_agents()

    # Get a specific agent definition
    agent = await client.get_agent("agent-123")

    # Report metrics back to control plane
    await client.report_metrics({
        "agent_id": "agent-123",
        "requests": 42,
        "avg_latency_ms": 320,
    })

    # Data plane heartbeat (keeps the control plane informed)
    await client.heartbeat(
        runtime_version="1.0.0",
        pod_count=3,
        queue_depth=12,
        last_seq=1500,
        runtime_token="your-runtime-token",
    )

3. Use the Memory Pipeline Directly

from vouchstone_sdk import MemoryPipeline, Entity, Skill

pipeline = MemoryPipeline(
    agent_id="agent-123",
    redis_url="redis://localhost:6379",
    vector_db_url="http://localhost:8002",
    graph_db_url="bolt://localhost:7687",
)
await pipeline.initialize()

# Before each turn — gather context from all 5 layers
context = await pipeline.prepare_context(
    session_id="sess-abc",
    user_input="How should we handle the CDC replication?"
)

# After each turn — persist to episodic + queue async extraction
result = await pipeline.process_turn(
    session_id="sess-abc",
    turn_number=1,
    user_input="How should we handle the CDC replication?",
    agent_response="I recommend Debezium for CDC with Kafka...",
    tools_used=["search", "knowledge_base"],
    tokens_in=120,
    tokens_out=85,
    latency_ms=1200,
    success=True,
)

# Upsert a semantic entity
await pipeline.semantic.upsert_entity("agent-123", Entity(
    id="e1", entity_type="technology", entity_key="Debezium",
    attributes={"category": "CDC", "use_case": "real-time replication"},
    confidence=0.95,
))

# Register a procedural skill
await pipeline.procedural.register_skill("agent-123", Skill(
    id="s1", name="cdc_setup",
    description="Set up CDC replication pipeline",
    steps=["Analyse source schema", "Configure Debezium connector", "Validate lag"],
    tools_required=["schema_analyzer", "kafka_admin"],
))

# Run meta-memory maintenance (decay, dedup, compress)
report = await pipeline.run_maintenance()

# End session (clears working memory)
await pipeline.end_session("sess-abc")
await pipeline.close()

---

Architecture

YOUR AGENT CODE (this SDK)
    |
    v
+-------------------+          +-------------------+
|   DATA PLANE      |  <--->   |   CONTROL PLANE   |
|                   |          |                   |
| Agent Runtime     |          | Dashboard (Next.js)|
| Working Memory    |  sync    | API (FastAPI)     |
|   (Redis)         |  ---->>  | PostgreSQL        |
| Semantic Memory   |          | Stripe Billing    |
|   (ChromaDB)      |          | ABAC / RACI       |
| Procedural Memory |          | Cost Governance   |
|   (Neo4j)         |          | Compliance Packs  |
+-------------------+          +-------------------+

5-Layer Memory Stack

Each agent has access to a biologically-inspired persistent memory architecture:

Layer	Name	Storage	Purpose	Lifecycle
1	Working	Redis	Current turn context window	Resets per session
2	Episodic	PostgreSQL	Turn-by-turn traces with importance scoring	Append-only, 90-day retention
3	Semantic	ChromaDB	Entity knowledge graph (people, tech, systems)	Upsert with merge on collision
4	Procedural	Neo4j	Learned skills as versioned DAG with success rates	Version-bumped on update
5	Meta	Control Plane	Decay, dedup, compress, archive, forget	Scheduled maintenance

---

SDK Components

Agent (Base Class)

Subclass Agent and implement run(). The base class handles:

• Session management (start_session, end_session)
• Memory context preparation (automatic before each turn)
• Post-turn persistence (automatic after each turn)
• Tool registration

AgentConfig

Field	Type	Default	Description
name	str	required	Agent display name
model	str	claude-sonnet-4-20250514	LLM model identifier
temperature	float	0.7	Sampling temperature
max_tokens	int	4096	Max output tokens
system_prompt	str	None	System prompt template
working_memory	bool	True	Enable Layer 1
semantic_memory	bool	True	Enable Layer 3
episodic_memory	bool	True	Enable Layer 2
procedural_memory	bool	True	Enable Layer 4
meta_memory	bool	True	Enable Layer 5
embedding_model	str	text-embedding-3-small	Embedding model for vector search

VouchstoneClient

Async HTTP client for the control plane API:

Method	Description
list_agents()	List all agents in your tenant
get_agent(id)	Fetch a specific agent definition
report_metrics(data)	Push runtime metrics to control plane
report_status(agent_id, status)	Report agent health status
heartbeat(...)	Data plane heartbeat (required for sync)
replay_ledger(entries)	Bulk replay audit ledger entries
fetch_agent_spec(...)	Fetch latest agent specifications

MemoryPipeline

Orchestrates all 5 memory layers:

Method	Description
prepare_context(session_id, input)	Gather context from all layers before a turn
process_turn(...)	Persist turn result to episodic + queue extraction
run_reflection(session_id)	Discover new skills from episodic traces
run_maintenance()	Run meta-memory (decay, dedup, compress)
get_snapshot()	Full memory snapshot across all layers
end_session(session_id)	Clear working memory for a session

Key Types

Type	Description
Message	Input message with content, role, metadata
AgentResponse	Response with content, decisions, tool calls, usage
MemoryContext	Full context from all 5 layers (passed to run())
EpisodicTrace	A single turn trace with importance score
Entity	A semantic entity (person, technology, system, etc.)
Skill	A procedural skill with steps, tools, success rate
HealthReport	Memory health stats and recommendations

---

Data Plane Sync

The SDK supports bidirectional sync with the Vouchstone control plane:

# Heartbeat — call every 30s to keep the control plane informed
await client.heartbeat(
    runtime_version="1.0.0",
    pod_count=3,
    queue_depth=12,
    last_seq=1500,
    runtime_token="your-runtime-token",
)

# Fetch latest agent specs (new deployments, config changes)
specs = await client.fetch_agent_spec(since=last_sync_timestamp)

# Replay audit entries from data plane to control plane
await client.replay_ledger(entries=[
    {"action": "agent_executed", "agent_id": "...", "timestamp": "..."},
])

---

Environment Variables

Variable	Required	Description
VOUCHSTONE_API_KEY	Yes	API key from your tenant
VOUCHSTONE_API_URL	No	Control plane URL (default: https://api.vouchstone.ai)
VOUCHSTONE_TENANT_ID	Yes	Your tenant identifier
REDIS_URL	No	Redis URL for working memory
CHROMADB_URL	No	ChromaDB URL for semantic memory
NEO4J_URL	No	Neo4j URL for procedural memory
OPENAI_API_KEY	No	For embedding generation (semantic layer)

---

Development

git clone https://github.com/GGChamp85/Vouchstone.git
cd Vouchstone/data-plane/sdk/python

pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Format
black vouchstone_sdk/

# Type check
mypy vouchstone_sdk/

---

License

Proprietary — Copyright (c) 2026 Vouchstone LLC. All Rights Reserved.