Python SDK¶
Beta — a thin, synchronous wrapper over the
REST API, built on requests.
Not on PyPI
There is no pip install venkai. Install from the repository —
Installation.
Client¶
from venkai.sdk.client import VenkaiClient
client = VenkaiClient(
api_key="vk_live_…", # required
project="billing-service", # optional default for every call
base_url="https://api.venkai.fr", # REQUIRED against the hosted API
timeout=10.0,
)
| Parameter | Default | Notes |
|---|---|---|
api_key |
— | Required. Empty raises VenkaiAuthError. |
project |
None |
Default project. Calls without one and without project= raise VenkaiAuthError. |
base_url |
http://127.0.0.1:8000 |
Points at localhost. Pass the hosted URL explicitly. |
timeout |
10.0 |
Seconds, per request. |
client.memory¶
remember(...)¶
client.memory.remember(
"We use Postgres, not MongoDB, because the ranking query needs joins.",
type="decision", # default "fact"
agent="planner", # default "sdk-agent"
importance=0.9, # default 0.5
confidence=0.95, # default 0.8
metadata={"ticket": "ARCH-14"},
project="billing-service", # optional if set on the client
)
Returns the created context dict. Wraps POST /api/memory.
remember() rebinds the client's default project
Passing project= also sets client.project, so later recall() /
decisions() calls inherit it. Convenient for single-project scripts,
surprising when one client is shared across projects — pass project=
explicitly everywhere in that case.
recall(...)¶
memories = client.memory.recall("which database did we pick", limit=5)
for m in memories:
print(round(m["score"], 3), m["type"], m["content"])
Wraps GET /api/context/{project}/relevant and returns the memories list
directly (not the envelope). limit defaults to 10.
decisions(...)¶
Wraps GET /api/context/{project}?type=decision and returns the decisions
array. Not ranked — chronological, newest first.
timeline(...)¶
for e in client.memory.timeline(limit=100):
print(e["timestamp"], e["agent"], e["type"], e["summary"])
Wraps GET /api/timeline/{project}. Entries are {timestamp, agent, type,
summary, context_id}, oldest first, with summary truncated to 140 characters.
client.security¶
Experimental
Wraps POST /api/security/analyze, which checks code against a reference
security knowledge graph. Treat the output as advisory: it has not been
validated against a labelled benchmark, and its response shape may change.
Errors¶
from venkai.sdk.client import VenkaiError, VenkaiAuthError
try:
client.memory.recall("anything")
except VenkaiAuthError:
... # bad/missing key, or no project resolved
except VenkaiError:
... # any other non-2xx from the API
VenkaiAuthError subclasses VenkaiError, so catch the specific one first.
What the SDK does not do¶
Set expectations before you build on it:
| Async | No. It is synchronous requests. Use asyncio.to_thread in async code. |
| Retries | None. No backoff, no 429 handling — wrap it yourself. |
| Connection pooling | Per-call requests functions; no shared Session. |
| Pagination helpers | None. Use the REST endpoint with offset for large lists. |
| Endpoint coverage | Memory + security only. Projects, versions, handoffs, export are REST-only. |
For anything beyond remember/recall, call the
REST API directly — it is a plain JSON API and the SDK gives you
nothing you would miss.
Complete example¶
import os
from venkai.sdk.client import VenkaiClient
client = VenkaiClient(
api_key=os.environ["VENKAI_API_KEY"],
base_url=os.environ.get("VENKAI_BASE_URL", "https://api.venkai.fr"),
project="billing-service",
)
client.memory.remember(
"Refunds over 500 EUR require a second approver.",
type="constraint", agent="policy-bot", importance=0.95,
)
task = "implement the refund endpoint"
context = client.memory.recall(task, limit=5)
prompt = "\n".join(f"- [{m['type']}] {m['content']}" for m in context)
print(f"Known constraints:\n{prompt}\n\nTask: {task}")