bee-sdk

Official Python client and MCP server for Bee by HEOSSI — The Progressive Quantum-Native Intelligence Engine, a governed multimodal intelligence platform from HEOSSI.

SDK version: 0.6.0 — adds MCP resources (bee://documents, bee://documents/{source}, bee://memory) on top of the documents (RAG) + personal-memory client methods and MCP tools, per-call/BEE_MODEL model-tier selection (bee-cell … bee-swarm; access follows your key's plan) and MCP tool annotations.

Includes a hosted Model Context Protocol server: pip install bee-sdk then run bee-mcp (stdio) to expose Bee's tools — bee_chat, bee_code, bee_security, bee_research, plus bee_documents_search/bee_documents_add (RAG, on rag-entitled plans) and bee_memory_search/bee_memory_add (personal memory, honours your opt-in) — to Claude Desktop, Cursor, VS Code, Zed, and Windsurf. It also serves resources: bee://documents (your knowledge-base manifest), bee://documents/{source} (one document's full text), and bee://memory (your remembered facts). See bee.heossi.com/docs/mcp.

Status: functional sync + async client (stdlib + optional httpx). The SDK targets the Bee /chat/completions API contract on production via the public gateway https://bee.heossi.com/bee — this is the default and it is where API-key auth, plan / per-tier allowance enforcement and usage metering happen. Do not point BEE_API_URL at the raw Modal app URL: that bypasses billing and a bee_sk_ key is rejected there (the backend only trusts Supabase JWTs / the static BEE_API_KEYS env, not customer-issued keys). Override BEE_API_URL only for a self-hosted Bee Enclave or staging.

Install

Install from PyPI (canonical):

pip install bee-sdk          # sync client (stdlib only — zero deps)
pip install bee-sdk[async]   # async client (adds httpx)

Install + quickstart on the marketing site: bee.heossi.com/docs/sdks.

Quick start

from bee_sdk import Bee

bee = Bee()  # reads BEE_API_URL + BEE_API_KEY from env
print(bee.chat("Explain Shor's algorithm at NISQ depth", domain="quantum"))

Quantum work is submitted as a durable Quantum Reasoning Lab product job. For customer-local execution, use quantum_local_select; for direct customer-owned provider execution, use execute_byopa_direct.

For a durable, inspectable Lab run:

job = bee.quantum_reasoning_create(
    prompt="Compare two fault-tolerant designs.",
    model="bee-hive",
    product="simulation_cloud",
)
detail = bee.quantum_reasoning_wait(job["id"], timeout=900)
print(detail["status"], detail.get("candidates"), detail.get("inference_receipt_id"))

Pass a stable idempotency_key when your application may retry creation. Reusing that key with different input returns 409; reusing it with the same input returns the original job. quantum_reasoning_jobs(cursor=..., limit=..., status=..., model=...) supports cursor pagination. Automatic execution retry is deliberately unavailable; ambiguous work must be reconciled. quantum_reasoning_remove cancels an eligible queued job or erases the content of a terminal job, subject to workspace role controls.

Lab jobs are tenant-scoped, encrypted at rest, and retained for 90 days. Real-QPU execution is explicitly metered and any classical fallback is returned as such.

Durable Lab jobs require an explicit product: simulation_cloud or managed_qpu. local_simulator and byopa_direct run in the customer's own environment and never enter Bee's hosted queue. byopa_managed remains gated until its provider-specific managed adapter is activated.

Streaming

for chunk in bee.chat_stream("Write a Rust fibonacci function", domain="programming"):
    print(chunk, end="", flush=True)

Async

import asyncio
from bee_sdk import AsyncBee

async def main():
    async with AsyncBee() as client:
        text = await client.chat("Audit this contract for re-entrancy", domain="blockchain")
        print(text)

asyncio.run(main())

Multi-turn

from bee_sdk import Bee, ChatMessage

bee = Bee()
resp = bee.chat_messages(
    [
        ChatMessage(role="system", content="You are a senior security auditor."),
        ChatMessage(role="user", content="Review this nginx config for hardening gaps:\n\n..."),
    ],
    domain="cybersecurity",
    max_tokens=1024,
)
print(resp.content)
print(resp.usage, resp.interaction_id)

Feedback loop

resp = bee.chat_messages([...], domain="ai")
if user_likes_answer:
    bee.feedback(resp.interaction_id, rating="up")

Documents (RAG) & personal memory

Both are tenant-scoped to your API key's account. Documents need a rag-entitled plan; memory is a per-user opt-in (default-on).

# Documents — add to your knowledge base, then search it
bee.documents_add("Q3 revenue grew 14% QoQ to S$2.1M.", source="q3-report")
hits = bee.documents_search("how did revenue change in Q3?", k=3)

# Personal memory — remember a fact, recall it later
bee.memories_add("Prefers TypeScript over JavaScript.", kind="preference")
mem = bee.memories_search("language preference")  # {enabled, memories: [...]}

Domains

The domain= parameter selects which LoRA adapter Bee routes through. Tier-1 domains:

Domain adapters are trained and served internally per domain; the model weights are private (the hosted gateway serves them — there is no public weight download). The domain you pass routes to the right adapter automatically.

Environment variables

Errors

from bee_sdk import BeeAPIError, RateLimitError, BeeError

try:
    bee.chat("...", domain="quantum")
except RateLimitError as e:        # 429 after retries
    ...
except BeeAPIError as e:           # other HTTP errors
    print(e.status, e.body)
except BeeError:                   # network / timeout
    ...

The sync client retries 429/5xx with exponential backoff (max 4 attempts).

Versioning

bee-sdk follows the Bee API surface in bee/server.py. Breaking API changes bump the minor version pre-1.0; the SDK is currently 0.1.1 and the API is v1.

License

Apache-2.0 © 2026 HEOSSI (Pte.) Ltd.