khive

Python client for the khive cognitive infrastructure API.

Preview release — 4 services: memory, communication, work, graph.

• Sync and async — both Khive and AsyncKhive clients
• Fully typed — Pydantic response models, complete type annotations
• Agent integration — tool schemas for OpenAI, Anthropic, and MCP
• Retry with backoff — exponential backoff with jitter, respects Retry-After

Installation

pip install khive

Quick Start

from khive import Khive

client = Khive(api_key="your-api-key")

# Store a memory
client.memory.remember(content="Quarterly review on March 15th")

# Recall memories by semantic search
results = client.memory.recall(query="quarterly review")

# Create a task
client.work.assign(title="Prepare slides", priority="p1")

# Send a message
client.communication.send(content="Slides ready", to="team")

# Knowledge graph
client.graph.create(type="person", name="Alice", role="engineer")
client.graph.link(source="node_1", target="node_2", relation="works_with")

Authentication

export KHIVE_API_KEY="your-api-key"

Or pass directly:

client = Khive(api_key="your-api-key")

Services

Memory (11 operations)

client.memory.recall(query="meeting notes", limit=5)
client.memory.remember(content="Sprint planning Monday 10am", tags=["recurring"])
client.memory.update(id="mem_123", importance=0.9)
client.memory.forget(id="mem_123")
client.memory.list(memory_type="episodic")
client.memory.feedback(memory_ids=["mem_123"], signal="positive")
client.memory.stats()

# Observational notes
from khive.types import NoteTarget
client.memory.note_create(
    targets=[NoteTarget(kind="memory", id="mem_123")],
    content="Frequently recalled in planning contexts",
)

Communication (18 operations)

client.communication.send(content="Status update", to="lambda:leo")
client.communication.list(status="unread")
client.communication.search(query="deployment")
client.communication.mark_read(message_id="msg_123")
client.communication.thread(thread_id="thread_123")
client.communication.discuss(discuss_action="start", topic="Architecture review")
client.communication.overview()

Work (25 operations)

client.work.next(limit=5)
client.work.assign(title="Fix login bug", priority="p1", energy="high")
client.work.complete(id="task_123")
client.work.schedule_add(title="Team standup", start="2026-05-01T10:00:00")
client.work.schedule_today()
client.work.areas()
client.work.project_create(area_id="area_123", name="Q1 Goals")
client.work.queue_decision(
    title="Choose database",
    options=["PostgreSQL", "SQLite", "DuckDB"],
)

Graph (7 operations)

client.graph.create(type="person", name="Alice", role="engineer")
client.graph.link(source="node_1", target="node_2", relation="works_with")
client.graph.search(query="engineers")
client.graph.neighbors(id="node_1", direction="outbound")
client.graph.get(id="node_1")
client.graph.unlink(source="node_1", target="node_2")
client.graph.delete(id="node_1")

Agent Integration

Generate tool schemas from the SDK's type system — no manual schema maintenance.

OpenAI

from khive import Khive

client = Khive()
tools = client.tools.to_openai_tools()

# Use with openai.chat.completions.create(tools=tools, ...)
# Then execute: client.tools.execute(tool_call.function.name, args)

Anthropic

tools = client.tools.to_anthropic_tools()
# Use with anthropic messages.create(tools=tools, ...)
# Then execute: client.tools.execute(block.name, block.input)

MCP

tools = client.tools.to_mcp_tools()

Schema-Only (no client needed)

from khive import to_openai_tools, list_tools

names = list_tools()  # All tool names
memory_tools = to_openai_tools(services=["memory"])

Async

import asyncio
from khive import AsyncKhive

async def main():
    async with AsyncKhive() as client:
        result = await client.memory.recall(query="recent decisions")
        await client.communication.send(content="Update", to="team")

asyncio.run(main())

Error Handling

from khive import Khive
from khive._errors import AuthenticationError, RateLimitError, NotFoundError

client = Khive()
try:
    client.memory.recall(query="test")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except NotFoundError:
    print("Resource not found")

Configuration

client = Khive(
    api_key="your-api-key",            # or KHIVE_API_KEY env var
    base_url="https://custom.api",      # or KHIVE_BASE_URL env var
    timeout=30.0,                       # request timeout (seconds)
    max_retries=3,                      # retries for transient errors
    default_headers={"X-Custom": "v"},  # additional headers
)