Quickstart
Install the Python host, declare your first capability, add persistence, verify, and serve over HTTP.
Zero mandatory dependencies.
pip install chp-coreWrap any function with @capability. The host handles evidence emission automatically.
from chp_core import LocalCapabilityHost, capability
host = LocalCapabilityHost("my-host")
@capability(
id="demo.greet",
version="1.0.0",
description="Return a greeting.",
)
def greet(name: str):
return {"message": f"Hello {name}"}
host.register(greet)
result = host.invoke(
"demo.greet",
{"name": "CHP"},
correlation_id="qs-001",
)
print(result.outcome) # → "success"
print(result.data) # → {"message": "Hello CHP"}
events = host.replay("qs-001")
# → [execution_started, execution_completed]setup_sqlite_capabilities() wires six SQLite-backed capability families in one call — evidence, state machine, event bus, ingestion, retrieval, knowledge graph, and incident tracking.
from chp_core import LocalCapabilityHost, SQLiteEvidenceStore
from chp_core import setup_sqlite_capabilities
store = SQLiteEvidenceStore(".chp/evidence.sqlite")
host = LocalCapabilityHost("my-agent", store=store)
# Wire 6 SQLite-backed capabilities in one call:
# state_machine, event_bus, ingestion, retrieval, graph, incident
managers = setup_sqlite_capabilities(host, base_dir=".chp")chp host verify smoke-tests the host and evidence store in under 1 second. Pass --store-dir to also verify persistent storage.
chp host verify
# → chp host is healthy — evidence recorded and replayed
# With a persistent store directory:
chp host verify --store-dir .chpExpose any host via a factory function. chp serve-http loads it and starts the server — GET /health, POST /invoke, GET /replay/{id}.
# In app.py — expose your host via a factory function
from chp_core import LocalCapabilityHost, setup_sqlite_capabilities
def create_host() -> LocalCapabilityHost:
host = LocalCapabilityHost("my-agent")
setup_sqlite_capabilities(host)
return hostchp serve-http --module app:create_host --port 8765
# Listening on http://127.0.0.1:8765
# Routes: GET /health, GET /host, GET /capabilities,
# POST /invoke, GET /replay/{id}, POST /replaySupply your own embedding function — any provider works. InMemoryVectorRetrievalCapability for dev, SQLiteVectorRetrievalCapability for prod. Every query emits retrieval_started/completed evidence.
from chp_core import (
LocalCapabilityHost,
InMemoryVectorRetrievalCapability,
register_retrieval_capability,
)
def embed(text: str) -> list[float]:
# Replace with openai.embeddings.create(), cohere.embed(), etc.
raise NotImplementedError
corpus = [
{"source_id": "doc-1", "title": "Getting Started", "content": "..."},
{"source_id": "doc-2", "title": "Evidence Model", "content": "..."},
]
host = LocalCapabilityHost("rag-agent")
register_retrieval_capability(
host, InMemoryVectorRetrievalCapability(embed, corpus)
)
result = host.invoke(
"retrieval.query",
{"query": "how does evidence work?", "top_k": 3},
correlation_id="rag-001",
)
for ref in result.data["source_refs"]:
print(f"[{ref['score']:.3f}] {ref['title']}")RemoteCapabilityHost mirrors the local invoke/replay API over HTTP. No new dependencies — just stdlib urllib.request.
# Cross-host composition — Agent B calls Agent A over HTTP
from chp_core import RemoteCapabilityHost
agent_b = RemoteCapabilityHost("http://127.0.0.1:8765")
result = agent_b.invoke(
"payments.transfer",
{"amount": 100.0, "to": "acct_456"},
correlation={"correlation_id": "cross-001"},
)
print(result.outcome) # → "success"
# Replay evidence on Agent A from Agent B
events = agent_b.replay("cross-001")
# → retrieval_started, retrieval_completed