Python SDK reference

from linkworld_sdk import (
    App,
    Context,
    InboundEnvelope,
    SDKRequest,
    AgentCallError,
    AgentResponse,
    MockAgent,
    MockSecrets,
    MockTools,
    ToolCallError,
    load_manifest,
    load_manifest_from_string,
)

App

app = App.from_manifest("linkworld.app.yaml")

Decorators

@app.tool("classify", description="...", scopes_required=["mail.send"])
async def classify(ctx: Context, text: str) -> dict:
    return {"category": "tax"}

@app.on_install
async def on_install(ctx: Context) -> None: ...

@app.on_uninstall
async def on_uninstall(ctx: Context) -> None: ...

@app.on_inbound
async def on_inbound(ctx: Context, env: InboundEnvelope) -> None: ...

@app.on_user_added
async def on_user_added(ctx: Context, user: dict) -> None: ...

@app.on_schedule("daily-summary")  # name from manifest.lifecycle.schedules
async def daily(ctx: Context) -> None: ...

# Custom HTTP routes (OAuth callbacks, webhooks, downloads)
@app.http_route("/oauth/callback", methods=["GET"], public=True)
async def oauth_cb(ctx: Context, request: SDKRequest) -> dict: ...

@app.http_route("/quotes/{quote_id}.pdf", methods=["GET"])
async def quote_pdf(ctx: Context, request: SDKRequest, quote_id: str) -> bytes: ...

Registration rules:

• Handler must be async def. Sync handlers raise TypeError at registration.
• scopes_required must be a subset of manifest.required_scopes. Mismatch raises ValueError.
• @app.on_X requires manifest.lifecycle.X = true. Mismatch raises ValueError.
• Schedule names must match manifest.lifecycle.schedules[].name.
• @app.http_route paths/methods must be declared in manifest.http_routes, and the public flag on the decorator must match the manifest entry.
• Double-registration of the same hook raises ValueError.

app.run()

app.run()                       # production: HTTP server on manifest.runtime.port
app.run(local=True)             # local: same shape + Mock backends

Production mode reads LINKWORLD_MCP_URL + LINKWORLD_MCP_TOKEN from container env (the platform injects them at provision time). Local mode swaps in MockTools + MockSecrets and exposes /__mock/tool and /__mock/secret endpoints for direct curl.

Context

Request-scoped helper passed to every handler.

ctx.tenant_id                            # str
ctx.tenant_slug                          # Optional[str] — e.g. 'ocean-tec'
ctx.app_id                               # str — your slug
ctx.event_type                           # 'tool' | 'on_inbound' | 'route' | 'public_route' | …
ctx.tools                                # ToolsApi    (sandboxed on public routes)
ctx.secrets                              # SecretsApi
ctx.agent                                # AgentApi    (sandboxed on public routes)
ctx.commitments                          # CommitmentsApi — see /developer/reference/commitments
ctx.kv                                   # KvApi — per-(tenant, app) JSONB store, /developer/reference/kv
ctx.config                               # dict[str, Any] — install-settings answers
ctx.user_id                              # Optional[str] — None on public routes
ctx.platform_origin                      # Optional[str] — e.g. 'https://app.linkworld.ai'
ctx.logger                               # logging.Logger — see note below

ctx.logger is stdlib, not structlog

The logger is a plain logging.Logger. Keyword-args other than the reserved stdlib ones (extra, exc_info, stack_info, …) crash the call. Two valid patterns:

# Use printf-style formatting:
ctx.logger.info("classified id=%s as %s", mail_id, category)

# Or pass structured fields via extra=:
ctx.logger.info("classified", extra={"mail_id": mail_id, "category": category})

What does NOT work (crashes at runtime):

ctx.logger.info("classified", mail_id=mail_id)   # TypeError

This is not a Linkworld decision — it’s how Python’s stdlib logger behaves. The eval suite catches the crash before deploy if your handlers go through it.

What does NOT exist on ctx:

• ctx.notify_user(...) — there is no notify primitive. To surface something in the user’s inbox, use ctx.commitments.add(...); the platform’s notification log mirrors commitments into the user’s inbox view.

ctx.public_callback_url(path)

Build the per-tenant URL for a public HTTP route declared in your manifest. Use this in on_install to register OAuth redirect URIs or webhook endpoints with a third-party.

@app.on_install
async def on_install(ctx: Context) -> None:
    redirect_uri = ctx.public_callback_url("/oauth/callback")
    # → "https://app.linkworld.ai/api/apps/<slug>/public/t/<tenant>/oauth/callback"
    await ctx.tools.call("integrations.register", uri=redirect_uri)

Raises RuntimeError if ctx.platform_origin or ctx.tenant_slug isn’t populated (e.g. unit-test stubs that built Context by hand). Raises ValueError for paths that don’t start with /, contain .., //, or NUL.

ctx.public_token_create(audience, *, ttl_seconds=900, claims=None)

Mint a short-lived signed token the platform later verifies on a public route — typical use: anti-CSRF token in an OAuth state param, or a one-shot HMAC the third-party callback presents back as proof the redirect originated from this install.

@app.on_install
async def on_install(ctx):
    state = ctx.public_token_create(
        "oauth-state",
        ttl_seconds=600,
        claims={"flow": "github"},
    )
    redirect = ctx.public_callback_url("/oauth/callback")
    return {"redirect_to": f"https://github.com/oauth?state={state}&redirect_uri={redirect}"}


@app.http_route("POST", "/oauth/callback", public=True)
async def callback(ctx, req):
    state = req.query["state"]
    claims = ctx.public_token_verify("oauth-state", state)
    if claims is None:
        return {"status": 401, "body": "invalid state"}
    # claims["flow"] == "github" — proceed with code-for-token exchange

Returns an opaque token (~120 chars). The matching public_token_verify(audience, token) returns the original claims dict on success or None if the token is expired, audience-mismatched, or signature-invalid. Both are scoped to this (tenant_id, app_id) — a token minted by tenant A can’t be verified by tenant B’s install.

App introspection

When you need to assert at runtime which handlers / tools registered (e.g. in on_install or in tests):

app.tool_names()      # ["classify_text", "summarize"]
app.schedules()       # ["inbox_sweep", "daily_digest"]
app.has_handler("on_inbound")  # True if @app.on_inbound was applied

These reflect what was registered by the import-time decorators — useful for evals, but the platform doesn’t enforce a check against them at deploy. Manifest validation is the source of truth.

ctx.tools.call(name, **kwargs)

result = await ctx.tools.call(
    "email_send",
    to="[email protected]",
    subject="Hello",
    body="…",
)

Returns the platform’s result dict. Raises ToolCallError on:

decision	Meaning
scope_denied	Your app or the tenant didn’t grant the required scope. exc.needed_scopes lists which
network_error	Couldn’t reach the platform (rare; container is in the same VM)
platform_error	5xx from the platform — transient
bad_response	Platform returned non-JSON
denied	Generic denial (rate-limit, gate decision, …)

ctx.secrets.get(key)

api_key = await ctx.secrets.get("OPENAI_KEY")
if api_key is None:
    raise RuntimeError("Tenant did not configure OPENAI_KEY")

Returns str | None. Never raises on auth/network failure — fail-closed. Key must match ^[A-Z][A-Z0-9_]{0,63}$ (validated SDK-side).

ctx.agent.ask(prompt, **opts)

Persona-aware LLM call as the manifest-declared specialist agent. Replaces the deprecated llm_complete / llm_vision tools.

res = await ctx.agent.ask(
    "Draft a LinkedIn post about flexible work.",
    agent="drafter",                      # manifest agent id; default if omitted
    response_format={                     # optional structured-output coercion
        "type": "object",
        "properties": {"text": {"type": "string"}},
    },
    images=[                              # optional vision input (base64 or data: URL)
        {"data": b64_screenshot, "media_type": "image/png"},
    ],
    max_tokens=4096,
)
res.text          # free-form answer (None when response_format forced tool_use)
res.data          # parsed structured output (None otherwise)
res.agent         # which agent slug actually handled it
res.input_tokens  # billed to the tenant
res.output_tokens

Raises AgentCallError on platform-side failure: app declares no agent, slug not found, no LLM provider connected, or upstream LLM error. Cost is attributed to the calling app via app_id_scope.

Platform tools for @app.on_install

Two platform tools that partner SDK apps typically call from their @app.on_install handler to replicate the privileged DB writes first-party apps used to do inline. Both are idempotent — safe to re-run on every re-activation.

doc_templates_install_defaults(app_id: str) -> dict

Install the calling app’s bundled default doc templates (Brief, Rechnung, Angebot, …) into the tenant’s linkworld_doc_templates table. Skips templates whose slug already exists for the tenant.

@app.on_install
async def install(ctx):
    result = await ctx.tools.call(
        "doc_templates_install_defaults",
        app_id="office-assistant",  # slug under src/apps/<slug>/doc_templates/
    )
    ctx.logger.info("templates_installed: count=%s", result["created_count"])

Returns {ok, created: [{slug, name}], created_count}. Scope: document.draft.

user_app_dashboard_create(...) -> dict

Create or fetch a dashboard tile (a linkworld_user_apps row) for the calling app. Idempotent on (tenant_id, slug-of-name) — re-runs return the existing row instead of letting the underlying service auto-suffix into duplicates.

@app.on_install
async def install(ctx):
    if ctx.user_id is None:
        return  # platform-initiated activation; skip per-user tile
    await ctx.tools.call(
        "user_app_dashboard_create",
        user_id=ctx.user_id,            # ctx user, NOT the synthetic
                                        # app-principal (tenant FK)
        name="Office Documents",
        description="Letters, invoices, minutes, quotes",
        icon="briefcase",
        ui_schema={"components": [...]},
    )

The skill validates that user_id is a member of the calling tenant before letting UserAppService.create write — partners can’t attribute dashboards to arbitrary UUIDs. Scope: files.write.

ctx.tools and ctx.agent on public routes

Public HTTP routes have NO authenticated tenant user — anyone on the internet hits them. So ctx.tools and ctx.agent are SANDBOXED: every call raises with a clear error. Read ctx.secrets instead for app-scoped values (OAuth state secrets, signing keys), then return the response. Calls from public routes that need user-scoped action should write a deferred work item (task_create etc.) and process it later from a private surface.

ToolCallError

class ToolCallError(Exception):
    tool_name: str            # tool that failed
    reason: str               # platform's error message
    decision: str             # see table — branch on this
    needed_scopes: list[str]  # populated when decision == "scope_denied"
    fix_hint: str             # imperative recovery step (may be empty)
    doc_link: str             # docs URL for the long version

decision values + their meaning are documented at Troubleshooting → ToolCallError. Recommended pattern:

try:
    result = await ctx.tools.call("email_search", received_after="-30m")
except ToolCallError as exc:
    ctx.logger.warning(
        "email_search failed: %s", exc.reason,
        extra={"decision": exc.decision, "fix_hint": exc.fix_hint},
    )
    if exc.decision == "approval_required":
        return  # tenant must approve — retry next tick
    if exc.decision == "scope_denied":
        await ctx.commitments.add(
            title="Add scope to manifest",
            body=exc.fix_hint or exc.reason,
        )
        return
    raise

The fix_hint and doc_link are populated by the platform when it recognizes the failure mode. For unknown / generic failures both are empty strings — handle the bare reason then.

AgentCallError

class AgentCallError(Exception):
    reason: str
    agent: Optional[str]

Raised by ctx.agent.ask when the platform refuses (no agent declared, slug not found, no LLM key, upstream error).

@app.http_route

Exposes raw HTTP endpoints the platform proxies to your container. The manifest must declare each route in http_routes: first.

@app.http_route("/oauth/callback", methods=["GET"], public=True)
async def stripe_cb(ctx: Context, request: SDKRequest) -> dict:
    code = request.query_params.get("code")
    state = request.query_params.get("state")
    # … verify state, exchange code, store tokens via ctx.secrets …
    return {
        "status": 200,
        "body": "<h1>Connected!</h1>",
        "headers": {"content-type": "text/html"},
    }

@app.http_route("/quotes/{quote_id}.pdf", methods=["GET"])
async def quote_pdf(ctx: Context, request: SDKRequest, quote_id: str):
    pdf_bytes = await render_pdf(ctx, quote_id)
    return {
        "status": 200,
        "body": pdf_bytes,
        "headers": {"content-type": "application/pdf"},
    }

Return shapes

Return	Status	Content-Type	Notes
dict (no status/body)	200	application/json	Returned as JSON body
bytes	200	application/octet-stream	Set explicit type via dict form
str	200	text/plain; charset=utf-8
None	204	—	No body
{"status": int, "body": ..., "headers": dict}	as set	as set	Explicit form

Public vs. private

Aspect	public: false (default)	public: true
Authentication	Tenant session cookie required	None
Tenant resolution	From session	From URL path: /api/apps/<slug>/public/t/<tenant>/...
ctx.user_id	Real user UUID	None
ctx.tools, ctx.agent	Available	Sandboxed — calls raise
Methods	GET/POST/PUT/DELETE/PATCH	GET only — others → 405
Timeout	60s	30s

Security floor (enforced by the platform — you don’t write this)

• 1MB inbound body cap, 10MB outbound — anything larger → 413/502
• Cookie and Authorization headers NEVER forwarded to your container
• Set-Cookie, Location, X-LW-* stripped from your response
• 1xx and 3xx status codes rewritten to 502 (no informational, no open-redirect)
• Forced response headers: Content-Security-Policy: sandbox; …, X-Content-Type-Options: nosniff, X-Frame-Options: DENY, Referrer-Policy: no-referrer
• Content-Type allowlist: text/html, text/plain, application/json, application/pdf, image/{png,jpeg,gif,webp,svg+xml}. Everything else (notably application/javascript) → 502
• Per-(tenant, app) inflight cap of 10 concurrent requests; over capacity → 503
• Path traversal, control chars, protocol-relative paths all rejected

SDKRequest

@dataclass(slots=True)
class SDKRequest:
    method: str               # 'GET' | 'POST' | …
    path: str                 # full path the partner declared
    query_params: dict[str, str]
    headers: dict[str, str]   # already filtered — no Cookie, no Authorization
    body: Optional[bytes]     # raw — partner parses
    client_ip: str

    def body_json(self) -> Any: ...

body_json() raises ValueError if the body is empty or invalid JSON. The SDK runtime catches handler exceptions and returns a generic 500 to the client, so try/except ValueError isn’t strictly necessary — just call body_json() and let the runtime handle malformed input.

MockTools / MockSecrets

For local dev:

tools = MockTools()
tools.register("email_send", lambda **kwargs: {"sent": True})

secrets = MockSecrets({"OPENAI_KEY": "sk-test"})

You don’t normally instantiate these directly — app.run(local=True) attaches them automatically. The /__mock/tool and /__mock/secret HTTP endpoints in local mode are sugar over the same classes.

InboundEnvelope

@dataclass
class InboundEnvelope:
    message_id: str
    channel: str        # 'email' | 'whatsapp' | 'webhook' | …
    from_: str          # alias for `from` (Python keyword)
    body: str
    subject: Optional[str]
    received_at: str    # ISO-8601 UTC
    attachment_ids: list[str]
    metadata: dict

Manifest helpers

load_manifest("linkworld.app.yaml")           # path → Manifest, or ManifestError
load_manifest_from_string("apiVersion: …")    # string → Manifest, or ManifestError

Both raise ManifestError (a wrapper around Pydantic’s ValidationError) on parse or schema failure.

The Pydantic models the SDK re-exports (importable from linkworld_sdk.manifest) cover every manifest block — useful when you want to programmatically build a manifest or run extra checks in your own tooling:

from linkworld_sdk.manifest import (
    Manifest, ManifestRuntime, ManifestTool, ManifestAgent,
    ManifestInstallSetting, ManifestHttpRoute, ManifestRagSlot,
    ManifestChrome, ManifestWorkflow, ManifestHeartbeat,
)

ManifestRagSlot lets you declare a tenant-RAG slot binding; see the RAGs reference for the full surface (5 tools, slot resolution, workspace-file ingestion).

Cross-app collaboration APIs

Four context surfaces for agent collaboration. Each has its own detailed reference page; quick one-liners here:

# Intra-app delegation (master → teammate)
result = await ctx.team.delegate(
    "researcher",
    "What are competitor X-followers numbers?",
    from_agent="cmo",       # required — caller's slug
)

# Cross-app RPC (Marketing → Sales)
result = await ctx.apps.invoke(
    "sales", "bdr",
    "Status of Acme deal?",
)

# Pub/sub event emit
result = await ctx.events.emit(
    "lead_qualified",
    {"lead_id": "L1", "score": 87},
)

# Commitments (follow-up tracking)
await ctx.commitments.add(
    from_agent="cmo",
    title="Q4 plan check-in",
    content="Recap launch metrics; confirm next sprint priority.",
    due_at_iso="2026-05-05T09:00:00Z",
    delivery_target={"kind": "room", "room_id": "<uuid>"},
)

# Group rooms
result = await ctx.rooms.post(
    "<room_id>",
    "@sales:bdr what's the Acme status?",
    from_agent="cmo",
)

Mock counterparts for unit tests:

from linkworld_sdk import (
    MockTools, MockSecrets, MockAgent,
    MockTeam, MockApps, MockEvents, MockRooms, MockCommitments,
)

Detailed reference:

• Heartbeats — declarative manifest schema
• Cross-app calls — ctx.team / ctx.apps / ctx.events
• Group rooms — ctx.rooms
• Commitments — ctx.commitments