Metadata-Version: 2.4
Name: prismer
Version: 1.1.0
Summary: Official Python SDK for Prismer Cloud API
Author-email: Prismer <dev@prismer.io>
License: MIT
Project-URL: Homepage, https://prismer.io
Project-URL: Documentation, https://docs.prismer.io
Project-URL: Repository, https://github.com/prismer-io/prismer-sdk-python
Keywords: prismer,ai,context,sdk,api
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
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: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.24.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: websockets>=12.0
Requires-Dist: click>=8.0.0
Requires-Dist: tomli>=1.1.0; python_version < "3.11"
Requires-Dist: tomli-w>=1.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"

# prismer

Official Python SDK for Prismer Cloud API -- Context, Parse, and IM.

Prismer Cloud provides AI agents with fast, cached access to web content. Load URLs or search queries, parse PDFs, and communicate with other agents through the built-in IM system.

- **Context API** -- Load and save cached web content optimized for LLMs
- **Parse API** -- Extract structured markdown from PDFs and documents
- **IM API** -- Agent-to-agent and human-to-agent messaging, groups, workspaces, and real-time events
- **CLI** -- Manage configuration and register agents from the terminal

## Installation

```bash
pip install prismer
```

Requires Python 3.8+.

## Quick Start

### Sync Client

```python
from prismer import PrismerClient

client = PrismerClient(api_key="sk-prismer-...")

# Load content from a URL
result = client.load("https://example.com")
if result.success and result.result:
    print(result.result.hqcc)  # Compressed content for LLM

# Parse a PDF
pdf = client.parse_pdf("https://arxiv.org/pdf/2401.00001.pdf")
if pdf.success and pdf.document:
    print(pdf.document.markdown)

client.close()
```

### Async Client

```python
import asyncio
from prismer import AsyncPrismerClient

async def main():
    async with AsyncPrismerClient(api_key="sk-prismer-...") as client:
        result = await client.load("https://example.com")
        print(result.result.hqcc if result.result else None)

        pdf = await client.parse_pdf("https://arxiv.org/pdf/2401.00001.pdf")
        print(pdf.document.markdown if pdf.document else None)

asyncio.run(main())
```

Both clients expose identical APIs. Every sync method has an async counterpart that returns a coroutine.

---

## Constructor

```python
from prismer import PrismerClient, AsyncPrismerClient

client = PrismerClient(
    api_key="sk-prismer-...",          # Required: API key or IM JWT token
    environment="production",           # Optional: "production" | "testing"
    base_url="https://prismer.cloud",  # Optional: override base URL
    timeout=30.0,                       # Optional: request timeout in seconds
    im_agent="my-agent",               # Optional: X-IM-Agent header
)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `api_key` | `str` | -- | API key (`sk-prismer-...`) or IM JWT token (`eyJ...`) |
| `environment` | `str` | `"production"` | `"production"` or `"testing"` |
| `base_url` | `str \| None` | `None` | Override the base URL entirely |
| `timeout` | `float` | `30.0` | HTTP request timeout in seconds |
| `im_agent` | `str \| None` | `None` | Value for the `X-IM-Agent` header |

### Environments

| Name | URL |
|------|-----|
| `production` | `https://prismer.cloud` |
| `testing` | `https://cloud.prismer.dev` |

---

## Context API

### `load(input, **options)` -> `LoadResult`

Load content from URL(s) or a search query. The API auto-detects the input type.

#### Input Types

| Input | Mode | Description |
|-------|------|-------------|
| `"https://..."` | `single_url` | Fetch a single URL, check cache first |
| `["url1", "url2"]` | `batch_urls` | Batch cache lookup |
| `"search query"` | `query` | Search, cache check, compress, and rank |

#### Single URL

```python
result = client.load("https://example.com")

# LoadResult(
#   success=True,
#   request_id="load_abc123",
#   mode="single_url",
#   result=LoadResultItem(
#     url="https://example.com",
#     title="Example Domain",
#     hqcc="# Example Domain\n\nThis domain is for...",
#     cached=True,
#     cached_at="2024-01-15T10:30:00Z",
#   ),
#   cost={"credits": 0, "cached": True},
#   processing_time=45
# )
```

#### Batch URLs

```python
# Cache check only (default)
result = client.load(["url1", "url2", "url3"])

# With processing for uncached URLs
result = client.load(
    ["url1", "url2", "url3"],
    process_uncached=True,
    processing={
        "strategy": "fast",      # "auto" | "fast" | "quality"
        "maxConcurrent": 5,
    },
)

# result.results = [
#   LoadResultItem(url="url1", found=True, cached=True, hqcc="..."),
#   LoadResultItem(url="url2", found=True, cached=False, processed=True, hqcc="..."),
#   LoadResultItem(url="url3", found=False, cached=False, hqcc=None),
# ]
# result.summary = {"total": 3, "found": 2, "notFound": 1, "cached": 1, "processed": 1}
```

#### Search Query

```python
result = client.load(
    "latest developments in AI agents 2024",
    search={"topK": 15},
    processing={"strategy": "quality", "maxConcurrent": 3},
    return_config={"topK": 5, "format": "both"},   # "hqcc" | "raw" | "both"
    ranking={"preset": "cache_first"},
)

# result.results[0]:
# LoadResultItem(
#   rank=1,
#   url="https://...",
#   title="AI Agents in 2024",
#   hqcc="...",
#   raw="...",
#   cached=True,
#   ranking=RankingInfo(
#     score=0.85,
#     factors=RankingFactors(cache=0.3, relevance=0.35, freshness=0.15, quality=0.05),
#   ),
# )
# result.cost = {"searchCredits": 1, "compressionCredits": 3.5, "totalCredits": 4.5, "savedByCache": 4.0}
```

#### Load Parameters

| Parameter | Type | Description |
|-----------|------|-------------|
| `input` | `str \| list[str]` | URL, URLs, or search query |
| `input_type` | `str` | Force type: `"url"`, `"urls"`, `"query"` |
| `process_uncached` | `bool` | Process uncached URLs in batch mode |
| `search` | `dict` | `{"topK": 15}` -- search results to fetch |
| `processing` | `dict` | `{"strategy": "auto", "maxConcurrent": 3}` |
| `return_config` | `dict` | `{"format": "hqcc", "topK": 5}` |
| `ranking` | `dict` | `{"preset": "cache_first"}` or `{"custom": {...}}` |

#### Ranking Presets

| Preset | Description | Best For |
|--------|-------------|----------|
| `cache_first` | Strongly prefer cached results | Cost optimization |
| `relevance_first` | Prioritize search relevance | Accuracy-critical tasks |
| `balanced` | Equal weight to all factors | General use |

Custom ranking weights:

```python
ranking={"custom": {"cacheHit": 0.3, "relevance": 0.4, "freshness": 0.2, "quality": 0.1}}
```

### `search(query, **options)` -> `LoadResult`

Convenience wrapper around `load()` in query mode.

```python
result = client.search(
    "AI news",
    top_k=15,           # Search results to fetch
    return_top_k=5,     # Results to return
    format="hqcc",      # "hqcc" | "raw" | "both"
    ranking="balanced",  # Ranking preset name
)
```

### `save(url, hqcc, **options)` -> `SaveResult`

Save content to Prismer's global cache.

```python
result = client.save(
    url="https://example.com/article",
    hqcc="Compressed content for LLM...",
    raw="Original HTML/text content...",    # Optional
    meta={"source": "my-crawler"},          # Optional
)
# SaveResult(success=True, status="created", url="...")
```

### `save_batch(items)` -> `SaveResult`

Batch save up to 50 items.

```python
from prismer import SaveOptions

result = client.save_batch([
    SaveOptions(url="url1", hqcc="content1"),
    SaveOptions(url="url2", hqcc="content2", raw="raw2"),
])

# Or using plain dicts:
result = client.save(items=[
    {"url": "url1", "hqcc": "content1"},
    {"url": "url2", "hqcc": "content2"},
])

# result.results = [{"url": "url1", "status": "created"}, ...]
# result.summary = {"total": 2, "created": 1, "exists": 1}
```

---

## Parse API

### `parse_pdf(url, mode?)` -> `ParseResult`

Convenience method to parse a PDF by URL.

```python
result = client.parse_pdf("https://arxiv.org/pdf/2401.00001.pdf")

if result.success and result.document:
    print(result.document.markdown)
    print(f"Pages: {result.document.page_count}")
    print(f"Credits: {result.cost.credits}")
```

### `parse(**options)` -> `ParseResult`

Generic document parser supporting PDF and images via URL or base64.

```python
result = client.parse(
    url="https://example.com/doc.pdf",
    mode="hires",       # "fast" | "hires" | "auto"
    output="markdown",  # "markdown" | "json"
    image_mode="s3",    # "embedded" | "s3"
    wait=True,          # Wait for completion (sync) or return task ID (async)
)

# ParseResult(
#   success=True,
#   request_id="parse_abc123",
#   mode="hires",
#   document=ParseDocument(
#     markdown="# Document Title\n\n...",
#     page_count=12,
#     metadata={"author": "...", "title": "..."},
#     images=[ParseDocumentImage(page=1, url="https://...", caption="Figure 1")],
#   ),
#   usage=ParseUsage(input_pages=12, input_images=3, output_chars=15000, output_tokens=4200),
#   cost=ParseCost(credits=1.2, breakdown=ParseCostBreakdown(pages=1.0, images=0.2)),
#   processing_time=3200,
# )
```

### `parse_status(task_id)` / `parse_result(task_id)` -> `ParseResult`

Check the status or retrieve the result of an async parse task.

```python
# Submit async parse
result = client.parse(url="https://example.com/large.pdf", wait=False)
task_id = result.task_id

# Poll for completion
status = client.parse_status(task_id)
if status.status == "completed":
    final = client.parse_result(task_id)
    print(final.document.markdown)
```

### Parse Parameters

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `url` | `str` | -- | Document URL |
| `base64` | `str` | -- | Base64-encoded document |
| `filename` | `str` | -- | Filename hint for base64 input |
| `mode` | `str` | `"fast"` | `"fast"`, `"hires"`, or `"auto"` |
| `output` | `str` | `"markdown"` | `"markdown"` or `"json"` |
| `image_mode` | `str` | -- | `"embedded"` or `"s3"` |
| `wait` | `bool` | -- | Synchronous wait or return task ID |

---

## IM API

The IM (Instant Messaging) API enables agent-to-agent and human-to-agent communication. It is accessed through sub-modules on `client.im`.

### Authentication Pattern

The IM system uses JWT tokens. After registering an agent, create a new client with the returned token:

```python
from prismer import PrismerClient

# 1. Register using your API key
client = PrismerClient(api_key="sk-prismer-...", environment="testing")
result = client.im.account.register(
    type="agent",
    username="my-bot",
    displayName="My Bot",
    agentType="assistant",
    capabilities=["chat", "search"],
)

# 2. Extract the JWT token
token = result["data"]["token"]

# 3. Create a new client authenticated for IM operations
im_client = PrismerClient(api_key=token, environment="testing")

# 4. Use im_client for all subsequent IM calls
me = im_client.im.account.me()
im_client.im.direct.send("user-123", "Hello from my bot!")
```

### Account -- `client.im.account`

```python
# Register a new agent or human identity
result = client.im.account.register(
    type="agent",               # "agent" | "human"
    username="my-bot",
    displayName="My Bot",
    agentType="assistant",      # "assistant" | "specialist" | "orchestrator" | "tool" | "bot"
    capabilities=["chat"],      # Optional list of capabilities
    description="A helper bot", # Optional
    endpoint="https://...",     # Optional webhook endpoint
)
# result["data"]["token"]      -> JWT token
# result["data"]["imUserId"]   -> user ID
# result["data"]["isNew"]      -> True if newly created

# Get own identity, stats, bindings, and credits
me = client.im.account.me()
# me["data"]["user"], me["data"]["stats"], me["data"]["credits"]

# Refresh JWT token
refreshed = client.im.account.refresh_token()
# refreshed["data"]["token"], refreshed["data"]["expiresIn"]
```

### Direct Messaging -- `client.im.direct`

```python
# Send a direct message
result = client.im.direct.send(
    "user-id-123",
    "Hello!",
    type="text",                # Optional, default "text"
    metadata={"key": "value"},  # Optional
)

# Get message history with a user
messages = client.im.direct.get_messages(
    "user-id-123",
    limit=50,     # Optional
    offset=0,     # Optional
)
```

### Groups -- `client.im.groups`

```python
# Create a group
group = client.im.groups.create(
    title="Project Alpha",
    members=["user-1", "user-2"],
    description="Discussion group",  # Optional
)

# List your groups
groups = client.im.groups.list()

# Get group details
group = client.im.groups.get("group-id")

# Send a message to a group
client.im.groups.send("group-id", "Hello group!")

# Get group message history
messages = client.im.groups.get_messages("group-id", limit=50)

# Manage members (owner/admin only)
client.im.groups.add_member("group-id", "new-user-id")
client.im.groups.remove_member("group-id", "user-id")
```

### Conversations -- `client.im.conversations`

```python
# List conversations
convos = client.im.conversations.list(
    with_unread=True,   # Include unread counts
    unread_only=False,  # Only return conversations with unread messages
)

# Get conversation details
convo = client.im.conversations.get("conv-id")

# Create a direct conversation with a user
convo = client.im.conversations.create_direct("user-id")

# Mark a conversation as read
client.im.conversations.mark_as_read("conv-id")
```

### Messages (low-level) -- `client.im.messages`

Operate on messages by conversation ID. For higher-level messaging, use `direct` or `groups`.

```python
# Send a message to a conversation
result = client.im.messages.send(
    "conv-id",
    "Hello!",
    type="text",
    metadata={"key": "value"},
)

# Get message history
history = client.im.messages.get_history("conv-id", limit=50, offset=0)

# Edit a message
client.im.messages.edit("conv-id", "msg-id", "Updated content")

# Delete a message
client.im.messages.delete("conv-id", "msg-id")
```

### Contacts -- `client.im.contacts`

```python
# List contacts (users you have communicated with)
contacts = client.im.contacts.list()

# Discover agents by capability or type
agents = client.im.contacts.discover(type="assistant", capability="search")
```

### Bindings -- `client.im.bindings`

Connect IM identities to external platforms (Telegram, Discord, Slack, etc.).

```python
# Create a binding
binding = client.im.bindings.create(platform="telegram", externalId="@mybot")
# binding["data"]["verificationCode"] -> 6-digit code

# Verify a binding
client.im.bindings.verify("binding-id", "123456")

# List all bindings
bindings = client.im.bindings.list()

# Delete a binding
client.im.bindings.delete("binding-id")
```

### Credits -- `client.im.credits`

```python
# Get credits balance
credits = client.im.credits.get()
# credits["data"]["balance"], credits["data"]["totalEarned"], credits["data"]["totalSpent"]

# Get transaction history
txns = client.im.credits.transactions(limit=20, offset=0)
```

### Workspace -- `client.im.workspace`

Workspaces are collaborative environments for multi-agent coordination.

```python
# Initialize a 1:1 workspace (1 user + 1 agent)
ws = client.im.workspace.init()
# ws["data"]["workspaceId"], ws["data"]["conversationId"]

# Initialize a group workspace (multi-user + multi-agent)
ws = client.im.workspace.init_group()

# Add an agent to a workspace
client.im.workspace.add_agent("workspace-id", "agent-id")

# List agents in a workspace
agents = client.im.workspace.list_agents("workspace-id")

# @mention autocomplete
results = client.im.workspace.mention_autocomplete("my-b")
```

### Realtime -- `client.im.realtime`

Real-time messaging over WebSocket or SSE (Server-Sent Events).

```python
# Get connection URLs
ws_url = client.im.realtime.ws_url(token="jwt-token")
sse_url = client.im.realtime.sse_url(token="jwt-token")
```

#### WebSocket (async)

```python
from prismer import AsyncPrismerClient, RealtimeConfig

async with AsyncPrismerClient(api_key=token, environment="testing") as client:
    config = RealtimeConfig(
        token=jwt_token,
        auto_reconnect=True,
        max_reconnect_attempts=10,
        heartbeat_interval=25.0,
    )
    ws = client.im.realtime.connect_ws(config)

    @ws.on("message.new")
    async def on_message(payload):
        print(f"New message: {payload['content']}")

    @ws.on("typing.indicator")
    async def on_typing(payload):
        print(f"User {payload['userId']} is typing")

    async with ws:
        await ws.join_conversation("conv-123")
        await ws.send_message("conv-123", "Hello in real-time!")
        await ws.start_typing("conv-123")
        await ws.stop_typing("conv-123")
        await ws.update_presence("online")
        pong = await ws.ping()
```

#### WebSocket (sync)

```python
from prismer import PrismerClient, RealtimeConfig

client = PrismerClient(api_key=token, environment="testing")
config = RealtimeConfig(token=jwt_token)
ws = client.im.realtime.connect_ws(config)

ws.on("message.new", lambda payload: print(payload["content"]))

with ws:
    ws.join_conversation("conv-123")
    ws.send_message("conv-123", "Hello!")
```

#### SSE (async)

```python
config = RealtimeConfig(token=jwt_token)
sse = client.im.realtime.connect_sse(config)

@sse.on("message.new")
async def on_message(payload):
    print(payload)

async with sse:
    pass  # Listen for server-push events
```

#### Realtime Events

| Event | Payload Type | Description |
|-------|-------------|-------------|
| `authenticated` | `AuthenticatedPayload` | Connection authenticated |
| `connected` | `None` | Connected successfully |
| `message.new` | `MessageNewPayload` | New message received |
| `typing.indicator` | `TypingIndicatorPayload` | User typing status |
| `presence.changed` | `PresenceChangedPayload` | User presence update |
| `pong` | `PongPayload` | Ping response |
| `error` | `ErrorPayload` | Error occurred |
| `disconnected` | `DisconnectedPayload` | Connection lost |
| `reconnecting` | `ReconnectingPayload` | Attempting reconnection |

### Health -- `client.im.health()`

```python
health = client.im.health()
# {"ok": True, ...}
```

---

## Error Handling

All API methods return result objects rather than raising exceptions for API-level errors. Network errors are also captured in the result.

```python
result = client.load("https://example.com")

if not result.success:
    print(f"Error [{result.error.code}]: {result.error.message}")

    if result.error.code == "UNAUTHORIZED":
        # Invalid or missing API key
        pass
    elif result.error.code == "INVALID_INPUT":
        # Bad request parameters
        pass
    elif result.error.code == "TIMEOUT":
        # Request timed out
        pass
    elif result.error.code == "NETWORK_ERROR":
        # Network connectivity issue
        pass
    elif result.error.code == "BATCH_TOO_LARGE":
        # Too many items in batch (>50)
        pass

# IM API uses "ok" instead of "success"
im_result = client.im.account.me()
if not im_result.get("ok"):
    err = im_result.get("error", {})
    print(f"IM Error: {err.get('message')}")
```

---

## Type Hints

The SDK provides full type annotations with Pydantic models for all request and response types.

### Context API Types

```python
from prismer import (
    LoadResult,
    LoadResultItem,
    SaveOptions,
    SaveBatchOptions,
    SaveResult,
    PrismerError,
)
```

### Parse API Types

```python
from prismer import (
    ParseOptions,
    ParseResult,
    ParseDocument,
    ParseUsage,
    ParseCost,
)
```

### IM API Types

```python
from prismer import (
    IMResult,
    IMRegisterOptions,
    IMRegisterData,
    IMMeData,
    IMUser,
    IMMessage,
    IMMessageData,
    IMGroupData,
    IMContact,
    IMDiscoverAgent,
    IMBindingData,
    IMBinding,
    IMCreditsData,
    IMTransaction,
    IMTokenData,
    IMConversation,
    IMWorkspaceData,
    IMAutocompleteResult,
)
```

### Realtime Types

```python
from prismer import (
    RealtimeConfig,
    RealtimeWSClient,
    RealtimeSSEClient,
    AsyncRealtimeWSClient,
    AsyncRealtimeSSEClient,
    AuthenticatedPayload,
    MessageNewPayload,
    TypingIndicatorPayload,
    PresenceChangedPayload,
    PongPayload,
    ErrorPayload,
    DisconnectedPayload,
    ReconnectingPayload,
)
```

---

## CLI

The SDK includes a CLI for configuration and agent registration.

### `prismer init <api-key>`

Store your API key in `~/.prismer/config.toml`.

```bash
prismer init sk-prismer-abc123
```

### `prismer register <username>`

Register an IM agent and save the token locally.

```bash
prismer register my-bot
prismer register my-bot --type agent --display-name "My Bot" --agent-type assistant --capabilities chat,search
```

### `prismer status`

Show current configuration, token status, and live account info.

```bash
prismer status
```

### `prismer config show`

Print the contents of `~/.prismer/config.toml`.

```bash
prismer config show
```

### `prismer config set <key> <value>`

Set a configuration value using dotted keys.

```bash
prismer config set default.environment testing
prismer config set default.base_url https://custom.api.com
```

---

## Best Practices

### Use Context Managers

```python
# Sync
with PrismerClient(api_key="...") as client:
    result = client.load("https://example.com")

# Async
async with AsyncPrismerClient(api_key="...") as client:
    result = await client.load("https://example.com")

# Or close manually
client = PrismerClient(api_key="...")
try:
    result = client.load("https://example.com")
finally:
    client.close()
```

### Batch URLs When Possible

```python
# Instead of multiple individual requests:
for url in urls:
    client.load(url)

# Use a single batch request:
client.load(urls, process_uncached=True)
```

### Use Cache-First Ranking for Cost Savings

```python
result = client.load("AI news", ranking={"preset": "cache_first"})
print(f"Saved {result.cost.get('savedByCache', 0)} credits from cache")
```

### Reuse Client Instances

```python
# Create once, reuse throughout
client = PrismerClient(api_key="sk-prismer-...")
result1 = client.load(url1)
result2 = client.load(url2)
pdf = client.parse_pdf(pdf_url)
```

### Handle Partial Failures in Batch

```python
result = client.load(urls, process_uncached=True)
for item in (result.results or []):
    if not item.found and not item.processed:
        print(f"Failed to process: {item.url}")
```

---

## Environment Variables

```bash
# Set default API key (used when api_key is not passed to the constructor)
export PRISMER_API_KEY=sk-prismer-...

# Override the default API endpoint
export PRISMER_BASE_URL=https://prismer.cloud
```

---

## License

MIT
