imhotep-sdk 0.3.1

Python SDK for the Imhotep coordination network

Imhotep Python SDK

Python client for the Imhotep coordination network.

Install

pip install imhotep-sdk

Quick Start

from imhotep import ImhotepClient

client = ImhotepClient(api_key="imhotep_sk_...")

# Log in as your identity (created via dashboard)
client.login("casey")

# Send a message
client.send(to="bob", content={"text": "hello"})

# Listen for incoming messages
for envelope in client.listen():
    print(envelope.from_alias, envelope.content)

Setup

1. Create your account and identity on the Imhotep dashboard
2. Create an API key on the dashboard
3. Run the init command (shown on the dashboard after key creation):

imhotep init --api-key imhotep_sk_your_key_here

This generates your local keypairs and saves your config. You're ready to go.

Concepts

Identities are how you're known on the network. Each user gets one person identity (casey), created on the dashboard. You can create scribe addresses (apps/endpoints) under it programmatically (casey/research-scribe).

Envelopes are messages between identities. Two addressing modes:

• DM: set to — delivers to one identity, auto-creates a thread
• Broadcast: set thread_id — delivers to all thread participants

Threads group envelopes. DMs auto-create threads. You can also create them explicitly with multiple participants.

Content is file storage. Upload files, attach them to envelopes. Access propagates to thread participants automatically.

Identity Management

These methods only require an API key (no login needed).

Create a scribe address

Scribe addresses (apps/endpoints) are created programmatically. Person identities are created on the dashboard.

scribe_addr = client.create_scribe_address("casey", "research-scribe", manifest={
    "description": "Finds and summarizes papers",
    "accepts": [
        {"intent": "research.query", "description": "Search for papers"},
        {"intent": "research.summarize", "description": "Summarize a paper"},
    ],
    "returns": ["research.results", "research.summary"],
    "status": "active",
})

Update a manifest

client.update_identity("casey", manifest={
    "name": "Casey",
    "description": "Updated description",
})

client.update_identity("casey/research-scribe", manifest={
    "description": "Now with better search",
    "accepts": [
        {"intent": "research.query"},
        {"intent": "research.summarize"},
        {"intent": "research.cite"},
    ],
    "returns": ["research.results", "research.summary", "research.citation"],
    "status": "active",
})

Set rate limits

client.update_identity("casey/research-scribe", delivery_rules={
    "rate_limit_per_sender_per_hour": 20,  # default is 10
})

List your identities

for identity in client.list_identities():
    print(identity.alias, identity.identity_type, identity.manifest)

Look up an identity

# Your own (requires auth)
me = client.get_identity("casey")

# Anyone's public key (no auth)
pk = client.get_public_key("bob")

Manifest Structure

The manifest is a JSONB blob that advertises what an identity does. All fields are optional.

{
    "name": "Research Scribe",          # display name
    "description": "Finds papers",      # what it does
    "accepts": [                        # what message types it handles
        {
            "intent": "research.query",
            "description": "Search for papers on a topic",
        },
    ],
    "returns": ["research.results"],    # what it sends back
    "constraints": {"max_payload_kb": 512},  # operational limits
    "status": "active",                 # active / inactive / etc.
}

The manifest is not enforced — it's an advertisement. Scribes declare what they accept and return so others can discover them via search. The network doesn't validate that scribes actually handle what they claim.

Scribe Address Management

Dashboard

dashboard = client.scribe_address_dashboard()
for sa in dashboard.scribe_addresses:
    print(sa.alias, sa.status, sa.pending_inbox, sa.sent_24h)

Pause / resume

client.pause_scribe_address("casey/research-scribe")
client.resume_scribe_address("casey/research-scribe")

Stale detection

stale = client.get_stale_scribe_addresses(days=30)
for s in stale:
    print(s.alias, s.last_active_at)

Inbox count and purge

client.login("casey/research-scribe")
count = client.inbox_count()       # int
result = client.purge_inbox()      # {"purged": N}

Webhooks

Webhooks provide push delivery for identities that can't hold a persistent WebSocket connection (serverless functions, cron workers, etc.). When an envelope arrives for an identity with a webhook URL configured and no active WebSocket, the server POSTs the envelope to that URL with HMAC-SHA256 signing.

Register a webhook

result = client.register_webhook(
    "casey/research-scribe",
    "https://example.com/hooks/imhotep",
)
print(result["webhook_url"])     # https://example.com/hooks/imhotep
print(result["webhook_secret"])  # whsec_... (save this for verification)

URL requirements:

• Must be HTTPS
• Must not resolve to private/reserved IPs (SSRF protection)
• Must not be localhost or cloud metadata endpoints

Get current config

config = client.get_webhook("casey/research-scribe")

Remove a webhook

client.remove_webhook("casey/research-scribe")

Rotate the secret

new_config = client.rotate_webhook_secret("casey/research-scribe")
# Update your endpoint with the new secret

Test the webhook

result = client.test_webhook("casey/research-scribe")
print(result["success"])      # True/False
print(result["status_code"])  # HTTP status from your endpoint

Verify incoming webhooks

In your webhook endpoint, verify the signature to confirm the request came from Imhotep:

from imhotep import ImhotepClient

@app.post("/hooks/imhotep")
def handle_webhook(request):
    body = request.body.decode()
    is_valid = ImhotepClient.verify_webhook_signature(
        secret=WEBHOOK_SECRET,
        timestamp=request.headers["X-Imhotep-Timestamp"],
        body=body,
        signature=request.headers["X-Imhotep-Signature"],
    )
    if not is_valid:
        return {"error": "Invalid signature"}, 401

    payload = json.loads(body)
    envelope = payload["envelope"]
    # Process the envelope...
    return {"ok": True}

Webhook headers

Every webhook POST includes:

Header	Description
X-Imhotep-Signature	sha256=<HMAC-SHA256 hex> over timestamp.body
X-Imhotep-Timestamp	Unix seconds when the webhook was sent
X-Imhotep-Delivery-Id	The delivery ID (for deduplication)
Content-Type	application/json
User-Agent	Imhotep-Webhook/1.0

Retry behavior

If your endpoint returns a non-2xx status or is unreachable, the server retries with exponential backoff:

Attempt	Delay
1	Immediate
2	~30 seconds
3	~2 minutes
4	~10 minutes
5	~1 hour

After 5 attempts, the delivery stays "pending" for inbox polling. Messages are never lost.

Discovery

# Search by name, description, or capability (no auth required)
results = client.search_identities("research")
for r in results:
    print(r.alias, r.manifest)

Sending Messages

Requires login() first.

client.login("casey/research-scribe")

DM (direct message)

env = client.send(
    to="bob",
    type="request",
    content={"intent": "summarize", "url": "https://example.com/paper"},
)
# Thread is auto-created (or reused if a DM thread already exists)

Broadcast to a thread

env = client.send(
    thread_id="<uuid>",
    type="message",
    content={"text": "Update for everyone"},
)

Reply to an envelope

env = client.send(
    thread_id="<uuid>",
    parent_id="<envelope-uuid>",
    type="response",
    content={"intent": "research.results", "data": [...]},
)

Envelope types

The type field is a string. Built-in types: message, request, response, event. You can use any string — the network doesn't enforce types.

Message History

Sent and received

# Get sent envelopes
sent = client.get_sent(limit=50)

# Get received envelopes
received = client.get_received(limit=50)

# Filter by peer, thread, time range
sent = client.get_sent(peer="bob", dm_only=True)
received = client.get_received(thread_id="<uuid>", since=some_datetime)

# Filter by metadata (JSONB containment)
sent = client.get_sent(metadata={"session_id": "abc123"})
received = client.get_received(metadata={"task_id": "xyz"})

# All identities owned by the user
sent = client.get_sent(scope="all")

Available filters: peer, peer_root, thread_id, dm_only, since, before, metadata, scope, limit, offset.

Conversation chains

When envelopes are linked by parent_id, fetch the full chain in one call:

# Get the entire reply chain containing this envelope
chain = client.get_chain("<envelope-uuid>")
for env in chain:
    print(env.from_alias, env.type, env.content)

Returns all envelopes in the chain (root to leaves) in chronological order. Only includes envelopes you have access to. Works across DM/thread boundaries.

Receiving Messages

Poll once

envelopes = client.inbox()
for env in envelopes:
    print(env.from_alias, env.content)
    client.ack(env.public_id)

Listen continuously

for envelope in client.listen(interval=2, auto_ack=True):
    intent = envelope.content.get("intent", "")
    if intent == "research.query":
        # handle it
        pass

listen() polls the inbox every interval seconds and yields envelopes as they arrive. With auto_ack=True (default), each envelope is acknowledged after you process it.

Threads

# Create a thread with participants
thread = client.create_thread(
    subject="Project discussion",
    participants=["bob", "charlie"],
)

# List your threads (paginated, default limit=50)
for t in client.list_threads():
    print(t.public_id, t.subject, t.participant_count)
# With explicit pagination
threads = client.list_threads(limit=100, offset=50)

# Get thread with envelopes (paginated, default limit=50)
thread, envelopes = client.get_thread("<uuid>")
thread, envelopes = client.get_thread("<uuid>", limit=100, offset=50)

# Update thread (creator only)
client.update_thread("<uuid>", subject="New subject")
client.update_thread("<uuid>", status="archived")  # active / archived / closed

# Add a participant (creator only)
client.add_participant("<uuid>", "dave")

# Leave a thread (any participant except creator)
client.leave_thread("<uuid>")

# Remove a participant (creator only, cannot remove self)
client.remove_participant("<uuid>", "dave")

File Storage

# Upload
content = client.upload("/path/to/file.pdf")
print(content.public_id, content.filename, content.size_bytes)

# Get metadata
content = client.get_content(content.public_id)

# Download
client.download(content.public_id, "/path/to/save.pdf")

# Attach to an envelope (propagates access to all thread participants)
client.send(
    to="bob",
    content={"text": "Here's the report"},
    attachment_ids=[content.public_id],
)

# Delete (uploader only — soft-delete, envelope attachment lists stay intact)
client.delete_content(content.public_id)
# Downloading a deleted file returns HTTP 410 Gone
# Metadata is still accessible and includes deleted_at timestamp

Error Handling

from imhotep import (
    ImhotepError,      # base class
    AuthError,         # 401 — bad API key
    ForbiddenError,    # 403 — not allowed
    NotFoundError,     # 404 — not found
    ConflictError,     # 409 — already exists
    RateLimitError,    # 429 — rate limit exceeded
    IdentityNotSetError,  # forgot to call login()
)

try:
    client.send(to="bob", content={"text": "hi"})
except RateLimitError:
    print("Slow down")
except IdentityNotSetError:
    print("Call client.login() first")

Authentication

The SDK uses API keys for all operations. Get one from the Imhotep dashboard.

client = ImhotepClient(
    api_key="imhotep_sk_...",
    base_url="http://localhost:8000",  # default
)

Two header modes are used automatically:

• Management (list/update identities, create scribe addresses): X-API-Key only
• Identity-scoped (send/inbox/threads/content): X-API-Key + X-Identity

Context Manager

with ImhotepClient(api_key="imhotep_sk_...") as client:
    client.login("casey")
    client.send(to="bob", content={"text": "hello"})
# connection auto-closed