StateWeave

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  LangGraph  │     │    MCP      │     │   CrewAI    │     │   AutoGen   │
│   Adapter   │     │   Adapter   │     │   Adapter   │     │   Adapter   │
└──────┬──────┘     └──────┬──────┘     └──────┬──────┘     └──────┬──────┘
       │                   │                   │                   │
       └───────────┬───────┴───────────┬───────┘                   │
                   │                   │                           │
                   ▼                   ▼                           ▼
            ┌──────────────────────────────────────────────────────────┐
            │              🧶  Universal Schema v1                     │
            │                                                          │
            │  conversation_history  ·  working_memory  ·  goal_tree   │
            │  tool_results_cache  ·  trust_parameters  ·  audit_trail │
            └──────────────────────────────────────────────────────────┘

$ python examples/quickstart.py

Exported: 2 messages
Framework: langgraph

Imported into MCP ✅
MCP has: 2 messages

Diff: 15 changes (framework tag + timestamps)

Done! The agent's memories migrated successfully. 🧶

pip install stateweave

from stateweave import LangGraphAdapter, StateWeaveSerializer

# Export from LangGraph
adapter = LangGraphAdapter(checkpointer=my_checkpointer)
payload = adapter.export_state("my-thread-id")

# Serialize for transport
serializer = StateWeaveSerializer()
raw_bytes = serializer.dumps(payload)

from stateweave import MCPAdapter

# Import into MCP
mcp_adapter = MCPAdapter()
mcp_adapter.import_state(payload)

# The agent resumes with its memories intact

from stateweave.middleware import auto_checkpoint

@auto_checkpoint(every_n_steps=5)
def run_agent(payload):
    # Your agent logic here — StateWeave checkpoints automatically
    return payload

from stateweave import EncryptionFacade, MigrationEngine

# Set up encrypted migration
key = EncryptionFacade.generate_key()
engine = MigrationEngine(
    encryption=EncryptionFacade(key)
)

# Full pipeline: export → validate → encrypt → transport
result = engine.export_state(
    adapter=langgraph_adapter,
    agent_id="my-agent",
    encrypt=True,
)

# Decrypt → validate → import on the other side
engine.import_state(
    adapter=mcp_adapter,
    encrypted_data=result.encrypted_data,
    nonce=result.nonce,
)

from stateweave import diff_payloads

diff = diff_payloads(state_before, state_after)
print(diff.to_report())
# ═══════════════════════════════════════════════
# 🔍 STATEWEAVE DIFF REPORT
# ═══════════════════════════════════════════════
#   Changes: 5 (+2 -1 ~2)
#   [working_memory]
#     + working_memory.new_task: 'research'
#     ~ working_memory.confidence: 0.7 → 0.95

StateWeavePayload(
    stateweave_version="0.3.0",
    source_framework="langgraph",
    exported_at=datetime,
    cognitive_state=CognitiveState(
        conversation_history=[...],   # Full message history
        working_memory={...},         # Current task state
        goal_tree={...},              # Active goals
        tool_results_cache={...},     # Cached tool outputs
        trust_parameters={...},       # Confidence scores
        long_term_memory={...},       # Persistent knowledge
        episodic_memory=[...],        # Past experiences
    ),
    metadata=AgentMetadata(
        agent_id="my-agent",
        access_policy="private",
    ),
    audit_trail=[...],               # Full operation history
    non_portable_warnings=[...],     # Explicit data loss docs
)

from stateweave import EncryptionFacade

# Generate a signing key pair
private_key, public_key = EncryptionFacade.generate_signing_keypair()

# Sign serialized payload
signature = EncryptionFacade.sign(payload_bytes, private_key)

# Verify on receipt
is_authentic = EncryptionFacade.verify(payload_bytes, signature, public_key)

from stateweave.core.delta import create_delta, apply_delta

# Create delta: only the differences
delta = create_delta(old_payload, new_payload)

# Apply delta on the receiver side
updated = apply_delta(base_payload, delta)

from stateweave.core.timetravel import CheckpointStore

store = CheckpointStore()

# Save a checkpoint
store.checkpoint(payload, label="before-experiment")

# View history
print(store.format_history("my-agent"))

# Roll back to a previous version
restored = store.rollback("my-agent", version=3)

# Branch from a checkpoint
store.branch("my-agent", version=3, new_agent_id="my-agent-experiment")

# Diff two versions
diff = store.diff_versions("my-agent", version_a=1, version_b=5)
print(diff.to_report())

from stateweave.a2a import A2ABridge

bridge = A2ABridge()

# Package state for A2A handoff
artifact = bridge.create_transfer_artifact(payload)

# Extract state from received A2A message
extracted = bridge.extract_payload(a2a_message_parts)

# Generate AgentCard skill for capability advertisement
caps = bridge.get_agent_capabilities()
skill = caps.to_agent_card_skill()

from stateweave.core.merge import merge_payloads, ConflictResolutionPolicy

result = merge_payloads(
    agent_a_state, agent_b_state,
    policy=ConflictResolutionPolicy.LAST_WRITER_WINS,
)
merged_payload = result.payload

from stateweave.adapters.base import StateWeaveAdapter
from stateweave.schema.v1 import StateWeavePayload, AgentInfo

class MyFrameworkAdapter(StateWeaveAdapter):
    @property
    def framework_name(self) -> str:
        return "my-framework"

    def export_state(self, agent_id: str, **kwargs) -> StateWeavePayload:
        # Translate your framework's state → Universal Schema
        ...

    def import_state(self, payload: StateWeavePayload, **kwargs):
        # Translate Universal Schema → your framework's state
        ...

    def list_agents(self) -> list[AgentInfo]:
        # Return available agents
        ...

# Show version and available adapters
stateweave version

# Dump the Universal Schema as JSON Schema
stateweave schema -o schema.json

# Validate a payload file
stateweave validate state.json

# Export/import/diff
stateweave export -f langgraph -a my-agent -o state.json
stateweave import -f mcp --payload state.json
stateweave diff before.json after.json

# Auto-detect source framework
stateweave detect state.json

# Scan environment for installed frameworks
stateweave scan

# Time travel — checkpoint, history, rollback
stateweave checkpoint state.json --label "before-experiment"
stateweave history my-agent
stateweave rollback my-agent 3 -o restored.json

# Run diagnostic health checks
stateweave doctor

# List all available adapters
stateweave adapters

# Scaffold a new adapter
stateweave generate-adapter my-framework --output-dir ./adapters

# Run UCE locally
python scripts/uce.py

# Run in CI mode (exit 1 on failure)
python scripts/uce.py --mode=CI --json

git clone https://github.com/GDWN-BLDR/stateweave.git
cd stateweave
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Run UCE
python scripts/uce.py

stateweave/
├── schema/        # Universal Schema (Pydantic models)
├── core/          # Engine (serializer, encryption, diff, delta, timetravel, environment, doctor)
├── adapters/      # Framework adapters (10 frameworks)
├── a2a/           # A2A protocol bridge
├── middleware/    # Auto-checkpoint middleware
├── playground/    # Interactive playground (REST API + UI)
├── registry/      # Schema registry (publish, search, discover)
├── templates/     # Project scaffolding (create-stateweave-agent)
├── mcp_server/    # MCP Server implementation
└── compliance/    # UCE scanners

[![StateWeave](https://img.shields.io/badge/state-StateWeave-7c3aed)](https://github.com/GDWN-BLDR/stateweave)