WormGPT App SDK

Official Python SDK for building applications, agents, and integrations on top of the WormGPT API. Sync + async clients, SSE streaming, automatic retries, typed responses, function tools, a batteries-included agent loop, webhook verification, pagination, a CLI, a Telegram bot runtime, and a JSON/SSE API server — in one dependency-light package (httpx).

from wormgpt import WormGPT

client = WormGPT()  # reads WORMGPT_API_KEY
reply = client.chat.create(messages="Explain exploit chaining in two lines.")
print(reply.content)

No HTML, no frontend build step, no browser required. Run the SDK directly, run a Telegram bot, or expose an HTTP API for other applications.

Install

From a checkout of this repository:

python -m pip install -e .

Or build and install the distribution artifacts:

python -m build
pip install dist/wormgpt_app_sdk-0.1.0-py3-none-any.whl

Requires Python 3.9+ and httpx.

Configuration

Argument	Environment variable	Default
api_key	WORMGPT_API_KEY	— (required)
base_url	WORMGPT_BASE_URL	https://wormgpt.app/v1
default_model	WORMGPT_MODEL	wormgpt-agent
timeout	WORMGPT_TIMEOUT	60 seconds
max_retries	WORMGPT_MAX_RETRIES	2

Explicit arguments always win over environment variables.

client = WormGPT(
    api_key="wg-...",
    base_url="https://wormgpt.app/v1",
    timeout=30,
    max_retries=3,
    organization="org_123",
    default_headers={"X-Trace": "abc"},
)

Use the client as a context manager to close the HTTP pool:

with WormGPT() as client:
    print(client.chat.create(messages="hi").content)

Chat completions

from wormgpt import WormGPT

client = WormGPT()

reply = client.chat.create(
    messages=[
        {"role": "system", "content": "You are terse."},
        {"role": "user", "content": "List three recon steps."},
    ],
    model="wormgpt-agent",
    temperature=0.3,
    max_tokens=256,
    stop=["\n\n"],
)

print(reply.content)
print(reply.finish_reason)
print(reply.usage.total_tokens)
print(reply.message.tool_calls)  # typed ToolCall objects, if any

messages accepts a string, a dict, a Message, or a list of those.

The hosted API is OpenAI-compatible, so the same call works with an OpenAI-style namespace:

reply = client.chat.completions.create(
    model="wormgpt-agent",
    messages=[{"role": "user", "content": "hello"}],
)

The official openai package also works against the same endpoint:

from openai import OpenAI

client = OpenAI(base_url="https://wormgpt.app/v1", api_key="wg-...")

Streaming (SSE)

stream = client.chat.stream(messages="Write a short poem about firewalls.")

with stream:
    for chunk in stream:
        print(chunk.content, end="", flush=True)

print("\nfinish:", chunk.finish_reason)

ChatStream is also a plain iterator; breaking out of the loop closes the connection. Server errors that arrive mid-stream raise StreamError.

Async

import asyncio
from wormgpt import AsyncWormGPT

async def main():
    async with AsyncWormGPT() as client:
        reply = await client.chat.create(messages="hello")
        print(reply.content)

        stream = await client.chat.stream(messages="count to three")
        async for chunk in stream:
            print(chunk.content, end="", flush=True)

asyncio.run(main())

Every resource has an async twin: client.chat, models, apps, sessions, files, usage. List endpoints return AsyncPage with await page.next_page() and async for item in page.auto_paging_iter().

Tools and the agent loop

@tool turns a typed Python function into a model-callable function and derives the JSON Schema from annotations, Annotated[...], Literal[...], dataclasses, enums, and docstrings.

from wormgpt import App, WormGPT

app = App(
    "recon-assistant",
    instructions="You are a precise recon assistant. Use tools when needed.",
    max_tool_rounds=6,
)

@app.tool
def resolve(host: str) -> str:
    """Resolve a hostname to an IP address."""
    import socket
    return socket.gethostbyname(host)

client = WormGPT()
run = app.run(client, "Where does example.com point?")
print(run.text)
print(run.steps, run.usage.total_tokens)

App.run keeps calling the model and executing tools until it gets a final answer, then returns an AppRun (text, messages, steps, usage, completions). App.arun is the async variant.

Need just the schema / dispatch pieces?

from wormgpt import ToolRegistry, tool

@tool(name="add", description="Add two integers")
def add(a: int, b: int) -> int:
    return a + b

registry = ToolRegistry([add])
result = registry.dispatch({
    "id": "call_1",
    "function": {"name": "add", "arguments": '{"a": 2, "b": 3}'},
})
print(result.content, result.is_error)  # 5 False

Tool exceptions are captured and returned to the model as error content by default; pass raise_on_error=True to dispatch if you want them to propagate.

Remote app registry

app = client.apps.create(
    name="recon-assistant",
    instructions="You are a precise recon assistant.",
    model="wormgpt-agent",
)
apps = client.apps.list(limit=20)
client.apps.update(app.id, description="updated")
client.apps.delete(app.id)

Sessions, files, usage

session = client.sessions.create(app_id=app.id, title="night shift")
history = client.sessions.history(session.id)
for message in history:
    print(message.role, message.content)

uploaded = client.files.upload("report.pdf", purpose="assistants")
raw = client.files.content(uploaded.id)

report = client.usage.retrieve(period="month")
print(report.total_tokens)

Pagination

page = client.models.list(limit=10)
for model in page:
    print(model.id)

# Iterate every page transparently.
for model in client.models.list().auto_paging_iter():
    print(model.id)

# Or step manually.
next_page = page.next_page()

Error handling

All errors derive from WormGPTError:

WormGPTError
├── ConfigurationError
├── APIConnectionError
│   └── APITimeoutError
├── APIStatusError
│   ├── BadRequestError           400
│   ├── AuthenticationError       401
│   ├── PermissionDeniedError     403
│   ├── NotFoundError             404
│   ├── ConflictError             409
│   ├── UnprocessableEntityError  422
│   ├── RateLimitError            429  (.retry_after)
│   └── ServerError               5xx
├── StreamError
├── WebhookSignatureError
└── MaxToolRoundsError

from wormgpt import RateLimitError, WormGPTError

try:
    client.chat.create(messages="hi")
except RateLimitError as exc:
    print("retry after", exc.retry_after)
except WormGPTError as exc:
    print(exc.status_code, exc.code, exc.request_id, exc.message)

Retries

The client retries timeouts, connection failures, 408, 409, 425, 429, and 5xx responses with exponential backoff and jitter, up to max_retries extra attempts. Retry-After is honored (capped at 60s). Streams are retried only before the first byte is delivered.

Webhooks

from wormgpt import construct_event, sign, verify

header = sign(raw_body, "whsec_...")            # "t=...,v1=..."

if verify(raw_body, header, "whsec_...", tolerance=300):
    event = construct_event(raw_body, header, "whsec_...")
    print(event.type, event.data)

Signatures are HMAC-SHA256 over f"{timestamp}." + body, compared in constant time. Stale timestamps are rejected by default (tolerance=300); pass tolerance=None to disable. Bare hex digests are also accepted.

CLI

export WORMGPT_API_KEY=wg-...

wormgpt chat "summarize this log" --stream
wormgpt chat "hello" --model wormgpt-agent --json
wormgpt models
wormgpt apps list
wormgpt usage --period month
wormgpt bot
wormgpt serve --host 127.0.0.1 --port 8000
wormgpt version

Telegram bot

Create a bot with @BotFather, then export both secrets and start the bot:

export WORMGPT_API_KEY=wg-...
export TELEGRAM_BOT_TOKEN=123456:ABC...
wormgpt bot

The bot uses Telegram's long-polling API directly (no extra dependency), keeps a small per-chat conversation history in memory, and answers with plain text split into Telegram-safe chunks.

Environment variable	Purpose
TELEGRAM_BOT_TOKEN	Bot token from BotFather (required)
TELEGRAM_ALLOWED_USERS	Comma-separated Telegram user IDs; everyone else is rejected
WORMGPT_SYSTEM_PROMPT	Default system prompt for every chat
WORMGPT_MODEL	Default model ID

Commands:

Command	Action
/start, /help	Show the command list
/reset	Clear this chat's history
/model / /model <id>	Show or switch the model
/models	List available models
/system <prompt>	Set a per-chat system prompt
/system reset	Clear the per-chat system prompt
/settings	Show model, history size, and prompt state
/id	Show chat and user IDs (useful for the allowlist)

Recommended lockdown:

export TELEGRAM_ALLOWED_USERS=123456789,987654321
wormgpt bot --model wormgpt-v5 --max-history 20

Run it from Python:

from wormgpt import run_bot

run_bot(token="123456:ABC...", model="wormgpt-agent")

run_bot reads WORMGPT_API_KEY, TELEGRAM_ALLOWED_USERS, and WORMGPT_SYSTEM_PROMPT from the environment when arguments are omitted. For custom clients:

from wormgpt import TelegramAPI, TelegramBot, WormGPT

with WormGPT() as client:
    bot = TelegramBot(client, TelegramAPI("123456:ABC..."), allowed_users={123456789})
    bot.run()

Conversation history is in memory only and disappears on restart. The bot token and WormGPT API key never leave the server.

HTTP API server

export WORMGPT_API_KEY=wg-...
wormgpt serve --host 127.0.0.1 --port 8000

The server exposes a small JSON/SSE API for bots, scripts, and other services:

Method	Path	Purpose
GET	/	Server name, SDK version, model, auth requirement, endpoints
GET	/api/models	Model list proxy
POST	/api/chat	Chat proxy; set "stream": true for SSE
POST	/chat	App endpoint with the tool loop
GET	/health	Health check

Non-streaming request:

curl -s http://127.0.0.1:8000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"messages": [{"role": "user", "content": "hello"}]}'

Streaming request (data: {"delta": "..."} frames, then data: [DONE]):

curl -N http://127.0.0.1:8000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"messages": [{"role": "user", "content": "count to five"}], "stream": true}'

POST /api/chat accepts messages, model, temperature, max_tokens, system, and stream. POST /chat accepts message (string) or messages, plus optional instructions, and runs the full agent/tool loop.

When binding to a network address, the server generates an eight-character access token and prints it. Send it as either header:

curl -H "Authorization: Bearer $TOKEN" http://127.0.0.1:8000/api/models
curl -H "X-Access-Token: $TOKEN" http://127.0.0.1:8000/api/models

Set a custom token with --token or WORMGPT_SERVER_TOKEN, or disable the gate on a trusted network with --no-auth. The /health and / routes stay public.

Run it from Python:

from wormgpt import App
from wormgpt.server import serve

app = App("echo", instructions="Repeat the user's intent briefly.")
serve(app, host="127.0.0.1", port=8000, access_token="change-me")

This is a development-grade threaded server, not a production ASGI stack. Put it behind a reverse proxy with TLS and rate limiting if you expose it beyond a trusted network.

Development

python -m pip install -e ".[dev]"
make test
make lint
make typecheck

Tests run against a local scriptable HTTP server; no API key or network access is required.

License

MIT © WormGPT