Pacerelle Python SDK

Python agent client for Pacerelle encrypted local agent relays.

For bounded, revocable permissions on local operations, see runtime permissions.

Use this SDK to connect a local Python process to Pacerelle, receive messages, reply to conversations, drive widgets, and return encrypted files or media.

Restarting an agent and replying to multiple sessions

Keep the same agent ID and store_root across restarts. With e2ee=True, the SQLite store persists the Signal identity and prekeys, conversation archive keys, decoded requests awaiting acceptance, accepted message IDs and each source's device. Existing opaque Signal snapshots migrate when next saved. The Python wrapper is specific to Python; do not copy it into another SDK's store.

Use reply_to_message_id=message.id on delayed replies. The SDK retains the source device, so a later request from a second browser does not redirect the first reply. Reading a duplicate accepted message acknowledges it again without calling the handler. A request interrupted before acceptance remains available on relay replay after reconnecting, including a request already decrypted.

The handler's successful return means the runtime has accepted responsibility. Persist the task before returning if work continues in the background. An acknowledgement does not prove that an action finished. The handler can run again after a failure or a crash before its acceptance commit: use the message ID as an idempotency key for effects outside the SDK. send_message reports submission to the socket; it does not wait for relay acknowledgement or guarantee exactly-once execution. Supervise connect() and reconnect after a network or handler error.

Clients advertising history-v1 can supply a conversation archive key inside an encrypted Signal delivery. Python unwraps it before the handler and attaches an AES-256-GCM archive record to subsequent replies. The relay receives ciphertext; the archive key stays with endpoints. The archive binds conversation, message, sender, key ID and epoch. This covers messages produced with the protocol; it does not recreate old missing plaintext or lost keys. Automatic key rotation is not provided by this Python client.

Use one running agent process per store directory. The SQLite file contains private keys and pending plaintext; protect the directory and its backups with the operating system. The current store does not encrypt the file or coordinate multiple writers at the agent level. Receipts are retained with the store and grow with accepted work. Losing or manually deleting the store loses its replay protection and Signal state. A peer identity notification or a failed decrypt does not reset unrelated sessions.

To compare the program's published Signal key through its local terminal:

client.publish_prekey_bundle()
print(client.get_verification_code())

Compare the full code with the agent verification dialog in Pacerelle. The getter does not create new prekeys, and connect() reuses this publication. A code sent through the conversation is not an independent comparison.

pip install --pre pacerelle

Alpha release: APIs may change before the first stable release. Production wheels bundle the native Signal runtime for the target platform.

Requirements

• Python 3.12 for the current alpha wheels.
• A supported platform wheel: Windows x64, Linux x64, Linux ARM64, or macOS ARM64.

Python 3.11 support is planned, but the current alpha release is tested and published for CPython 3.12 only.

Before You Run

Create an agent in Pacerelle. The confirmation panel shows both Identifiant de l'agent and Jeton d'authentification. Use Copier la configuration .env to copy the required variables.

export PACERELLE_AGENT_ID="agent-id"
export PACERELLE_AGENT_TOKEN="agent-token"

On Windows PowerShell:

$env:PACERELLE_AGENT_ID = "agent-id"
$env:PACERELLE_AGENT_TOKEN = "agent-token"

Published packages connect to the Pacerelle API by default. Local source builds default to http://localhost:8080 for development.

Agent Connect

Third-party Python applications can install an agent only after the user approves the request in Pacerelle. Keep the PKCE verifier, state, installation token, and runtime token on the application server.

from pacerelle import (
    begin_agent_connect,
    exchange_agent_connect_code,
    request_agent_runtime_token,
)

pending = begin_agent_connect(
    connect_key=user_supplied_connect_key,
    client_id=pacerelle_client_id,
    redirect_uri="https://your-app.example/pacerelle/callback",
    agent_name="Research assistant",
)
# Redirect the user to pending.authorization_url and verify pending.state.

installation = exchange_agent_connect_code(
    code=callback_code,
    code_verifier=pending.code_verifier,
    client_id=pacerelle_client_id,
    redirect_uri="https://your-app.example/pacerelle/callback",
)
runtime = request_agent_runtime_token(
    installation_token=installation.installation_token,
)

Minimal Echo Agent

import asyncio
import os

from pacerelle import AgentGatewayClient

client = AgentGatewayClient(
    token=os.environ["PACERELLE_AGENT_TOKEN"],
    agent_id=os.environ["PACERELLE_AGENT_ID"],
    e2ee=True,
)


async def handle(message, agent):
    await agent.send_message(
        conversation_id=message.conversation_id,
        to=message.from_id,
        reply_to_message_id=message.id,
        text=f"Received: {message.text}",
    )


client.on_message(handle)
asyncio.run(client.connect())

Incoming Messages

The handler receives an AgentMessage:

async def handle(message, agent):
    print(message.id)
    print(message.conversation_id)
    print(message.from_id)
    print(message.text)
    print(message.attachments)
    print(message.widget_response)

Use message.from_id as the to value when replying to the user.

Sending Messages And Replies

Send a normal message:

await agent.send_message(
    conversation_id=message.conversation_id,
    to=message.from_id,
    text="I can help with that.",
)

Reply to a specific user message:

await agent.send_message(
    conversation_id=message.conversation_id,
    to=message.from_id,
    reply_to_message_id=message.id,
    text="Replying to your last message.",
)

Running Your Own Agent Logic

async def handle(message, agent):
    result = await run_my_agent(message.text)

    await agent.send_message(
        conversation_id=message.conversation_id,
        to=message.from_id,
        reply_to_message_id=message.id,
        text=result,
    )

Widgets

Widgets are sent as encrypted conversation messages. Each method returns the widget id. User answers arrive later as message.widget_response.

Confirm

await agent.send_confirm_widget(
    conversation_id=message.conversation_id,
    to=message.from_id,
    widget_id="confirm-delete",
    title="Delete file?",
    body="This cannot be undone.",
    danger=True,
    labels={"yes": "Delete", "no": "Cancel"},
)

Handle the answer:

if message.widget_response and message.widget_response.ref == "confirm-delete":
    if message.widget_response.cancelled:
        return
    if message.widget_response.value is True:
        await agent.send_message(
            conversation_id=message.conversation_id,
            to=message.from_id,
            text="Confirmed.",
        )

Choice

await agent.send_choice_widget(
    conversation_id=message.conversation_id,
    to=message.from_id,
    widget_id="choose-format",
    title="Choose a format",
    options=[
        {"id": "pdf", "label": "PDF"},
        {"id": "csv", "label": "CSV"},
    ],
    multi=False,
)

Permission

await agent.send_permission_widget(
    conversation_id=message.conversation_id,
    to=message.from_id,
    widget_id="permission-files",
    title="Allow file access?",
    body="The agent needs access to selected files.",
    scopes=["once", "session"],
)

Form

await agent.send_form_widget(
    conversation_id=message.conversation_id,
    to=message.from_id,
    widget_id="profile-form",
    title="Complete profile",
    submitLabel="Save",
    fields=[
        {"name": "email", "label": "Email", "type": "email", "required": True},
        {"name": "notes", "label": "Notes", "type": "textarea"},
    ],
)

Progress

progress_id = await agent.send_progress_widget(
    conversation_id=message.conversation_id,
    to=message.from_id,
    widget_id="import-progress",
    title="Importing files",
    value=10,
    max=100,
    cancellable=True,
)

Update it:

await agent.send_widget_update(
    conversation_id=message.conversation_id,
    to=message.from_id,
    ref=progress_id,
    spec={"value": 65, "body": "Almost done"},
)

File Picker

await agent.send_file_picker_widget(
    conversation_id=message.conversation_id,
    to=message.from_id,
    widget_id="pick-files",
    title="Choose files",
    multiple=True,
    accept=[".pdf", "image/*"],
    max_files=5,
)

Date And Time

await agent.send_datetime_widget(
    conversation_id=message.conversation_id,
    to=message.from_id,
    widget_id="schedule",
    title="Pick a meeting time",
    mode="datetime",
    min="2026-05-21T09:00:00",
)

Files And Media

send_file and send_media encrypt bytes locally with AES-GCM, upload only ciphertext to /agent/blobs, then send the attachment key, IV, file name, type and dimensions inside the E2EE message payload. The relay only sees a generic name and the size.

Download and decrypt a file received from a user:

async def handle(message, agent):
    for attachment in message.attachments or []:
        data = await agent.download_attachment(attachment)
        print(attachment.name, attachment.mime, len(data))

await agent.send_file(
    conversation_id=message.conversation_id,
    to=message.from_id,
    reply_to_message_id=message.id,
    text="Here is the report.",
    name="report.txt",
    mime="text/plain",
    data=b"private report",
)

Media adds optional dimensions or duration:

await agent.send_media(
    conversation_id=message.conversation_id,
    to=message.from_id,
    text="Preview attached.",
    name="chart.png",
    mime="image/png",
    data=png_bytes,
    width=1200,
    height=800,
)

Encryption

When e2ee=True, the SDK encrypts and decrypts messages locally before they leave your machine. On connect, the client publishes the agent pre-key bundle, establishes encrypted sessions for conversations, and keeps message contents opaque to the relay.

Use e2ee=False only for local debugging or non-encrypted transports.

Groups

Group messages are encrypted once with a group key that human members distribute to each member over Signal. The client fetches it from /agent/group-keys, stores it with the rest of its state, and uses it for group targets (group:<conversation_id>). reply() answers in the group automatically:

async def handle(message, agent):
    if message.text.strip() == "!status":
        await agent.reply(message, "All green")

Collective votes use response_mode="collective" on send_choice_widget or send_confirm_widget.

MCP

This package is the Python SDK for building agents. The MCP server is distributed separately:

npx -y @pacerelle/mcp-server