rossum-agent-client 1.1.0

Python client for Rossum Agent API - AI-powered document processing assistant

Rossum Agent Client

Python client for Rossum Agent API - AI-powered document processing assistant.

Installation

uv pip install rossum-agent-client

CLI Usage

The package provides a rossum-agent-client command for single-turn interactions:

# Execute a prompt directly (single-turn conversation)
rossum-agent-client -x "List all queues"

# Read prompt from a markdown file (single-turn conversation)
rossum-agent-client -r prompt.md

# With explicit configuration
rossum-agent-client \
    --agent-api-url https://your-agent-api.example.com \
    --rossum-api-base-url https://elis.rossum.ai/api/v1 \
    --rossum-api-token your-token \
    -x "List all queues"

# Use read-write mode
rossum-agent-client --mcp-mode read-write -x "Create a new queue"

Files created by the agent (via write_file tool) are automatically downloaded and saved to the current directory.

Environment Variables

Variable	Description
ROSSUM_AGENT_API_URL	Agent API URL
ROSSUM_API_BASE_URL	Rossum API base URL
ROSSUM_API_TOKEN	Rossum API authentication token
ROSSUM_MCP_MODE	MCP mode: read-only (default) or read-write

Quick Start

from rossum_agent_client import RossumAgentClient

# Initialize client
client = RossumAgentClient(
    agent_api_url="https://your-agent-api.example.com",
    rossum_api_base_url="https://elis.rossum.ai/api/v1",
    token="your-rossum-api-token",
)

# Create a chat session
chat = client.create_chat(mcp_mode="read-only")
print(f"Created chat: {chat.chat_id}")

# Send a message and stream the response
last_content = ""
for event in client.send_message_stream(chat.chat_id, "List all queues"):
    if event.type == "tool_start":
        print(f"\n[Tool] {event.tool_name}")
    elif event.type == "final_answer":
        # Print only new content (events contain cumulative text)
        print(event.content[len(last_content):], end="", flush=True)
        last_content = event.content
    elif event.type == "done":
        print(f"\n({event.input_tokens} in, {event.output_tokens} out)")

Async Usage

import asyncio
from rossum_agent_client import AsyncRossumAgentClient

async def main():
    async with AsyncRossumAgentClient(
        agent_api_url="https://your-agent-api.example.com",
        rossum_api_base_url="https://elis.rossum.ai/api/v1",
        token="your-rossum-api-token",
    ) as client:
        # Create chat
        chat = await client.create_chat()

        # Stream response
        async for event in client.send_message_stream(chat.chat_id, "Hello!"):
            print(event)

asyncio.run(main())

API Reference

Client Initialization

RossumAgentClient(
    agent_api_url: str,         # Agent API URL
    rossum_api_base_url: str,   # Rossum API base URL (e.g., https://elis.rossum.ai/api/v1)
    token: str,                 # Rossum API authentication token
    timeout: float = 300.0      # Request timeout in seconds
)

Methods

Health Check

health = client.health_check()
# Returns: HealthResponse(status="healthy", redis_connected=True, version="1.0.0dev")

Chat Management

# Create a new chat
chat = client.create_chat(mcp_mode="read-only")  # or "read-write"

# List all chats
chats = client.list_chats(limit=50, offset=0)

# Get chat details
chat_detail = client.get_chat(chat_id)

# Delete a chat
result = client.delete_chat(chat_id)

Messages

# Send message and stream response (recommended)
last_content = ""
for event in client.send_message_stream(chat_id, "Your message"):
    match event.type:
        case "tool_start":
            print(f"\n[Tool] {event.tool_name}")
        case "final_answer":
            # Print only new content (events contain cumulative text)
            print(event.content[len(last_content):], end="", flush=True)
            last_content = event.content
        case "done":
            print()  # Final newline

# Send message with images
from rossum_agent_client.models import ImageContent
import base64

with open("invoice.png", "rb") as f:
    image_data = base64.b64encode(f.read()).decode()

images = [ImageContent(media_type="image/png", data=image_data)]
for event in client.send_message_stream(chat_id, "Extract data from this invoice", images=images):
    print(event)

# Send message with PDF documents
from rossum_agent_client.models import DocumentContent

with open("invoice.pdf", "rb") as f:
    pdf_data = base64.b64encode(f.read()).decode()

documents = [DocumentContent(media_type="application/pdf", data=pdf_data, filename="invoice.pdf")]
for event in client.send_message_stream(chat_id, "Process this document", documents=documents):
    print(event)

Files

# List files in a chat
files = client.list_files(chat_id)

# Download a file
content = client.download_file(chat_id, "report.csv")

SSE Event Types

When streaming messages, you'll receive these event types:

Event Type	Field	Description
thinking	event.content	Agent's reasoning
intermediate	event.content	Partial response
tool_start	event.tool_name	Tool being called
tool_result	event.result	Tool output
final_answer	event.content	Final response
error	event.content	Error message
file_created	event.filename, event.url	Generated file
sub_agent_progress	event.tool_name, event.iteration	Sub-agent status
sub_agent_text	event.text	Sub-agent output
done	event.input_tokens, event.output_tokens, event.token_usage_breakdown	Token usage with optional breakdown by main agent vs sub-agents

Note: Text events (thinking, intermediate, final_answer) contain cumulative content - each event includes all previous text plus new tokens. To display live progress, print only the delta: event.content[len(last_content):].

Models

All request/response models are available in rossum_agent_client.models:

from rossum_agent_client.models import (
    # Requests
    CreateChatRequest,
    MessageRequest,
    ImageContent,
    DocumentContent,

    # Responses
    ChatResponse,
    ChatDetail,
    ChatListResponse,
    ChatSummary,
    HealthResponse,
    DeleteResponse,
    FileListResponse,
    FileInfo,

    # Events
    StepEvent,
    StreamDoneEvent,
    FileCreatedEvent,
    SubAgentProgressEvent,
    SubAgentTextEvent,
    Message,
    TextContent,

    # Token usage
    TokenUsageBySource,
    SubAgentTokenUsageDetail,
    TokenUsageBreakdown
)

Error Handling

from rossum_agent_client.exceptions import (
    RossumAgentError,      # Base exception
    AuthenticationError,   # 401 errors
    NotFoundError,         # 404 errors
    RateLimitError,        # 429 errors
    ValidationError,       # 422 errors
    ServerError            # 5xx errors
)

try:
    chat = client.get_chat("non-existent-id")
except NotFoundError as e:
    print(f"Chat not found: {e}")
except RateLimitError as e:
    print(f"Rate limited, retry after: {e.retry_after}")

OpenAPI Specification

The full OpenAPI specification is available at openapi.json in this package, or at runtime:

# From running server
GET /api/openapi.json

# Interactive docs
GET /api/docs      # Swagger UI
GET /api/redoc     # ReDoc

License

MIT License - see LICENSE for details.

Resources

• Full Documentation
• Rossum Agent README
• Rossum API Documentation
• Main Repository