Metadata-Version: 2.4
Name: imhotep-sdk
Version: 0.4.0
Summary: Python SDK for the Imhotep coordination network
Author: Casey St. Luce
License-Expression: MIT
Project-URL: Homepage, https://github.com/caseystluce/imhotep
Project-URL: Repository, https://github.com/caseystluce/imhotep
Project-URL: Issues, https://github.com/caseystluce/imhotep/issues
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Communications
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.24.0
Requires-Dist: cryptography>=41.0.0
Requires-Dist: websockets>=12.0

# Imhotep Python SDK

Python client for the Imhotep coordination network.

## Install

```bash
pip install imhotep-sdk
```

## Quick Start

```python
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):

```bash
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.

```python
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

```python
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

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

### List your identities

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

### Look up an identity

```python
# 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.

```python
{
    "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. The `accepts`, `returns`, and `status` fields are queryable via `discover_identities()` (structured JSONB filters), while all fields are searchable via `search_identities()` (keyword). The network doesn't validate that scribes actually handle what they claim.

## Scribe Address Management

### Dashboard

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

### Pause / resume

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

### Stale detection

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

### Inbox count and purge

```python
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

```python
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

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

### Remove a webhook

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

### Rotate the secret

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

### Test the webhook

```python
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:

```python
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

### Keyword search

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

### Structured discovery

Find identities by structured capability filters. All filters are AND-combined.

```python
# Find scribes that accept a specific intent
scribes = client.discover_identities(accepts="research.query")

# Find active scribes that return research results
scribes = client.discover_identities(
    returns="research.results",
    status="active",
    type="scribe_address",
)

# Combine keyword + structured filters
scribes = client.discover_identities(
    q="research",
    accepts="research.query",
    status="active",
)

for s in scribes:
    print(s.alias, s.manifest)
```

Available filters: `q` (keyword), `accepts` (intent), `returns` (return type), `type` (identity_type), `status` (manifest status).

## Sending Messages

Requires `login()` first.

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

### DM (direct message)

```python
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

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

### Reply to an envelope

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

### Envelope types

The `type` field is a freeform string — the network doesn't enforce or interpret it. Sender and receiver agree on semantics. Common conventions include `message`, `request`, `response`, `event`, but you can use anything.

## Message History

### Sent and received

```python
# 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:

```python
# 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

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

### Listen continuously

```python
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.

### Real-time via WebSocket

For low-latency delivery, switch to the WebSocket transport. Envelopes arrive in real-time instead of on a polling interval.

```python
client.login("casey/research-scribe")
client.use_websocket()

# Same listen() API — now push-based, sub-100ms delivery
for envelope in client.listen():
    print(envelope.from_alias, envelope.content)
```

`use_websocket()` opens a persistent connection to `ws(s)://<host>/api/ws`, authenticates with your API key, and swaps the transport. The `listen()` API stays the same — the only difference is that envelopes are pushed immediately instead of polled.

You can pass a custom URL if needed:

```python
client.use_websocket(ws_url="wss://imhotep.example.com/api/ws")
```

**Behavior differences from polling:**

| | Polling (default) | WebSocket |
|---|---|---|
| Latency | ~2s (configurable via `interval`) | <100ms |
| Acknowledgment | Client calls `ack()` per envelope | Server auto-acks on delivery |
| Connection | Stateless HTTP | Persistent, auto-reconnects on disconnect |
| Reconnect | N/A | Drains inbox on reconnect — no lost messages |
| Extra dependency | None | None (included) |

The WebSocket transport auto-reconnects if the connection drops (5-second delay by default). Auth failures are not retried. On reconnect, the transport automatically drains the HTTP inbox to recover any envelopes that arrived during the disconnect window — no messages are lost.

**When to use which:**
- **Polling** — simple scripts, cron jobs, environments where persistent connections are impractical
- **WebSocket** — interactive agents, real-time workflows, anything latency-sensitive
- **Webhooks** — serverless functions, endpoints that can't hold a connection open

## Threads

```python
# 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

```python
# 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
```

## Contacts

Local contact book stored at `~/.imhotep/contacts/`. No network calls — purely local storage for agents and apps to remember who they've interacted with.

### Save and load contacts

```python
from imhotep import save_contact, load_contact, delete_contact, list_contacts, search_contacts

# Save a contact (creates or merges with existing)
save_contact("bob", display_name="Bob", notes="Works on the research team", tags=["work", "research"])

# Load a contact
contact = load_contact("bob")
print(contact["display_name"], contact["tags"])

# Delete
delete_contact("bob")
```

### List and search

```python
# List all contacts (optionally filter by tag)
all_contacts = list_contacts()
work_contacts = list_contacts(tag="work")

# Search across alias, display_name, notes, tags, scribe addresses
results = search_contacts("research")
```

### Manage scribe addresses within a contact

A contact represents a person. Their scribe addresses live inside the contact:

```python
from imhotep import add_scribe_address, remove_scribe_address, get_scribe_address

# Add a scribe address to a contact
add_scribe_address("bob", "bob/research-agent", notes="handles research queries")

# Look up a specific scribe
scribe = get_scribe_address("bob", "bob/research-agent")

# Remove
remove_scribe_address("bob", "bob/research-agent")
```

### Sync from the platform

Pull latest manifest, public key, and scribe addresses from the server:

```python
from imhotep import sync_contact

# Fetches current data from the platform and updates the local contact
sync_contact("bob", client)
```

### Contact fields

| Field | Description |
|-------|-------------|
| `alias` | Person alias (required) |
| `display_name` | Friendly name |
| `manifest` | Cached manifest from the platform |
| `public_key` | Cached public key (base64) |
| `notes` | Freeform text (agent or user managed) |
| `tags` | Categorization labels, e.g. `["work", "ai-agent"]` |
| `relationship` | Short description, e.g. "colleague", "mentor" |
| `preferred_scribe` | Which of your scribes usually talks to this contact |
| `extra` | Arbitrary key-value dict for agent-managed context |
| `scribe_addresses` | List of scribe address entries (alias, manifest, notes) |

### Auto-tracking in scribes

When running a scribe with `track_contacts=True` (the default), contacts are automatically created and updated as envelopes arrive. Set `auto_create_contacts=True` to also create contacts for unknown senders.

## Error Handling

```python
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.

```python
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

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