eolaswork

Official Python SDK for the EolasWork agentic platform.

Install

pip install eolaswork

Python 3.11+. Sync + async clients ship together.

Two ways to run agents

The model used for each run is fixed by the chosen role or team's manifest (default_model). There is no per-call model override today; if your tenant has multiple models wired up, change the role's manifest to switch.

Quickstart: chat with attachments

from eolaswork import Client

client = Client(api_key="nxa_...")        # or set EOLASWORK_API_KEY

# 1. Discover what's available
me     = client.account.whoami()
roles  = client.roles.list()              # pick a role.slug for the run
models = client.models.list()             # each Model has .id (UUID), .display_name, .provider

# 2. Create the task (bound to an agent + first message in one call)
task = client.tasks.create(
    role="research-analyst",                # role.slug from client.roles.list()
    first_message="Build a board-ready summary from the Excel I'll upload.",
    subject="Q2 board prep",                # optional label
)

# 3. Attach files to the task (the agent sees them on next turns)
client.files.upload(task.id, "./Q2_sales.xlsx")

# 4. Send the follow-up turn that asks the agent to use the file
task = client.tasks.send_message(
    task.id,
    text="The Excel is uploaded - produce the board summary now.",
)

# 5. Wait for the latest run to finish (or stream events live)
final = client.runs.wait(task.last_run_id, timeout=300)
print(final.status, final.output)

# Streaming alternative:
for ev in client.runs.stream(task.last_run_id):
    print(ev.kind, ev.payload)

Quickstart: standalone run (no task / files)

run = client.runs.create(
    prompt="Summarise today's INGEST team Slack channel.",
    role="research-analyst",                       # role.slug
    webhook_url="https://my-app.example.com/eolaswork/hook",  # optional
)

# Three ways to handle completion:
final = client.runs.wait(run.id, timeout=300)               # blocks
# OR live SSE:
for ev in client.runs.stream(run.id): print(ev.kind, ev.payload)
# OR don't wait - your webhook receiver gets the HMAC-signed POST.

print(final.status, final.output)

Async

import asyncio
from eolaswork import AsyncClient

async def main():
    async with AsyncClient(api_key="nxa_...") as client:
        task = await client.tasks.create(
            role="research-analyst",
            first_message="Summarise the latest project status.",
        )
        final = await client.runs.wait(task.last_run_id, timeout=300)
        print(final.status, final.output)

asyncio.run(main())

Compaction (many runs on one task)

When you keep sending runs to the same task, the message history grows and eventually crowds the model's context window. compact() compresses the older turns into a dense recap (the same compression that fires automatically when the budget is exceeded), so later runs stay efficient. Call it periodically between runs on a long-lived task:

for batch in batches:
    task = client.tasks.send_message(task.id, text=batch)
    client.runs.wait(task.last_run_id, timeout=300)

res = client.tasks.compact(task.id)
# {"compacted": True, "tokens_before": ..., "tokens_after": ...}
# or {"compacted": False, "reason": "Nothing to compact yet"}

client.tasks.compactions(task.id)              # history of compactions
# client.tasks.compaction(task.id, compaction_id)  # one record (summary + tokens)

Webhook receiver

from eolaswork.webhooks import verify_signature

@app.post("/eolaswork/hook")
def hook(request):
    payload = verify_signature(
        raw_body=request.body,
        signature_header=request.headers["X-EolasWork-Signature"],
        secret=os.environ["EOLASWORK_WEBHOOK_SECRET"],
    )
    print(payload.run_id, payload.status, payload.output)
    return "", 204

Configuration

Explicit constructor args win over env vars:

client = Client(api_key="...", base_url="https://eolaswork.your-co.com",
                timeout=120.0, max_retries=5)

Running behind a corporate or notebook-environment proxy

If your environment has HTTPS_PROXY / HTTP_PROXY set globally (common on Kaggle, Colab, corporate notebooks, VPN'd workstations) and the proxy blocks eolaswork.com with a 403 Forbidden, pass trust_env=False to bypass it for SDK calls only:

client = Client(api_key="nxa_...", trust_env=False)

Or point at a specific proxy:

client = Client(api_key="nxa_...", proxy="http://corp-proxy:8080")

Resource surface

Async variants share the exact same surface on AsyncClient -- every method is awaitable.

Memory (knowledge graph)

client.memory is a per-user knowledge graph - entities (each with free-form observations) and the directed relations between them - shared across web, desktop, and the API. It's the same graph the agent reads and writes during runs, so you can seed it, inspect it, or keep it in sync with your own systems.

from eolaswork import Client

ew = Client()  # reads EOLASWORK_API_KEY

ew.memory.create_entities([
    {"name": "Alice", "entityType": "person", "observations": ["prefers one-page briefs"]},
    {"name": "Acme", "entityType": "company", "observations": ["based in Dublin"]},
])
ew.memory.create_relations([
    {"from": "Alice", "to": "Acme", "relationType": "works_at"},
])
ew.memory.add_observations([
    {"entityName": "Acme", "contents": ["FY26 revenue target EUR 4m"]},
])

graph = ew.memory.read_graph()           # {"entities": [...], "relations": [...]}
hits = ew.memory.search("dublin")        # entities matching + relations among them
ew.memory.open(["Alice", "Acme"])        # specific entities by name

ew.memory.delete_entities(["Acme"])      # also drops its observations + relations

Shapes: entity {"name", "entityType", "observations": [...]}, relation {"from", "to", "relationType"}. All operations are scoped to the authenticated user. Narrative per-user notes (preferences, working style) live separately on client.account.get_instructions() / set_instructions().

License

MIT.