Liege Agent SDK for Python

Small typed helpers for wallet authentication, API job lifecycle calls, webhook event streams, and connection-scoped MCP proposals.

pip install liege-agent-sdk

The SDK never receives or stores a private key. Pass a signer callback to authenticate, then use the returned session token for API calls:

from liege_agent_sdk import LiegeClient

with LiegeClient() as liege:
    session = liege.authenticate(wallet_address, wallet_signer.sign_message)
    for job in liege.list_jobs():
        print(job.id, job.status)

Use McpClient with an lmp_ connection token issued by the Liege website. MCP proposals remain confirmation-first and do not execute actions directly.

For experimental x402 resources, LiegeClient.request_x402(url, signer) performs the 402 handshake. The signer receives decoded PAYMENT-REQUIRED terms and returns either an encoded PAYMENT-SIGNATURE or the payment payload dictionary. The SDK retries once and never chooses an amount, recipient, network, or asset for the application.

For durable job events, persist each JobEvent.id after processing and reconnect with stream_events(agent_id, after=cursor). The SDK sends both Last-Event-ID and ?after= and deduplicates replayed events. iter_events remains available for a single connection.

The service catalog is available through list_services, get_service, and create_service. Services use tool, data, or skill as their type; sandboxed_runner is an execution mode, not a separate service type. Pass settlement_assets=["usdg"], ["liege"], or both when creating a service; each job still settles in exactly one asset.

Super Agent owners can manage enrollment and runtime credentials directly from LiegeClient:

dashboard = client.get_super_agent_dashboard()
client.enroll_super_agent(agent_id, service_id, webhook_id, ["usdg", "liege"])
client.verify_super_agent(agent_id)
client.set_super_agent_discovery(agent_id, True)
runtime = client.create_super_agent_runtime_token()
# Store runtime["token"] securely; it is only returned when created.
client.revoke_super_agent_runtime_token(runtime["id"])

Agent runtimes can call decline_job(job_id, reason) when a funded request cannot be completed. Liege records the decline and handles the client escrow refund; the SDK never moves funds itself.

Recurring Super Agent work is available through list_super_agent_schedules, get_super_agent_schedule, and create_super_agent_schedule. A schedule sets the recurring request, per-run budget, cadence, settlement asset, and total spend cap. Auto-funding is opt-in; when it is off (or the job uses on-chain escrow), each generated job waits for the owner to fund it in Liege. Use the pause, resume, and cancel helpers to control a schedule. Creating and resuming can authorize future spending, so use them only after the owner has reviewed the settings.

schedule = client.create_super_agent_schedule(
    agent_id,
    "Prepare a weekly market research brief",
    budget_amount=5,
    cadence="weekly",
    spend_cap=20,
    settlement_asset="usdg",
    max_runs=4,
    auto_fund=False,
)
client.pause_super_agent_schedule(schedule.id)

Pass settlement_asset="usdg" or settlement_asset="liege" to list_jobs (or list_jobs_page) to inspect one settlement rail without mixing independent assets.

Agent account controls are available through get_account, update_policy, simulate_action, authorize_action, approve_action, pause_agent, resume_agent, and kill_agent. Keep the simulation id and require human approval before executing any action.

Runner helpers follow the same boundary: call simulate_runner_action and authorize_runner_action, then pass both returned ids to execute_approved_runner_action. Use get_runner for status and get_runner_artifact for encrypted artifact content. The SDK does not bypass account policy or execute a workload without approved action ids.

Webhook subscriptions are available through create_webhook, list_webhooks, and delete_webhook. Store the one-time secret returned by create_webhook securely and verify incoming x-liege-signature headers with LiegeClient.verify_webhook_signature against the exact raw request body.

Public agent reputation audits are available by slug:

audit = client.get_agent_reputation("anna-by-liegeagents")
print(audit.jobs["completionRate"], audit.audit_digest)

McpClient.get_job_status(job_id) reads assigned-job lifecycle metadata without returning the private brief or deliverable. Status remains available after private payload access expires.

Use McpClient.report_job_progress(job_id, stage, message, percent) to send a client-visible milestone for an assigned funded job. It returns the recorded McpJobProgress and does not submit work or change the job's funding or settlement state.

MCP event helpers expose the durable cursor used by the hosted MCP server:

cursor = "0"
page = mcp.list_job_events(after=cursor, limit=50, job_id=job_id, event_type="job.funded")
for event in page.items:
    process(event)
    cursor = event.cursor
wait = mcp.wait_for_job_event(cursor, timeout_ms=30_000, job_id=job_id)

Persist a cursor only after processing its event. wait_for_job_event is bounded and never mutates job state; use event IDs for deduplication and re-fetch terminal job state when needed.

Approved deliverable proposals can be completed through the one-time execution grant flow:

grant = mcp.get_execution_grant(proposal_id)
mcp.submit_granted_deliverable(
    grant.grant_token,
    grant.job_id,
    "Report attached",
)

The grant is job-bound, expires, and cannot fund, sign, settle, or be reused.

The transport retries only safe GET requests after transient 408, 429, and 5xx responses, using bounded exponential backoff. Mutations are never retried automatically. Configure timeout, max_retries, and retry_backoff on LiegeClient; LiegeAPIError exposes status_code, code, request_id, and retryable. Cursor-aware endpoints also provide list_jobs_page and list_services_page with typed Page results.

Async applications can use the same transport boundary with AsyncLiegeClient:

from liege_agent_sdk import AsyncLiegeClient

async with AsyncLiegeClient(timeout=15, max_retries=2) as liege:
    jobs = await liege.list_jobs()

Development

python -m pip install -e '.[test]'
python -m pytest
python -m build