linkworld-sdk

Python SDK for building apps on the Linkworld Open App Platform.

Status: 1.1.x. Production runtime online. Single-agent and multi-agent apps can be deployed and run against tenants. M3.11 adds custom HTTP routes (@app.http_route) for OAuth callbacks, webhooks, and downloadable artefacts.

Install

pip install linkworld-sdk

60-second tour

from linkworld_sdk import App

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

@app.on_inbound
async def handle(ctx, env):
    await ctx.tools.call(
        "email_send",
        to="ops@example.com",
        subject="Inbound!",
        body=f"From {env.user_id}: {env.message_text}",
    )

if __name__ == "__main__":
    app.run()

The matching manifest:

apiVersion: linkworld.ai/v2
app_id: my-app
version: 0.1.0
name: My App
required_scopes: [mail.send]
runtime:
  image: ghcr.io/your-org/my-app:0.1.0
lifecycle:
  on_inbound: true

Local development

LINKWORLD_LOCAL=1 python main.py

This starts a trigger HTTP server you can curl to simulate any platform event. Tools are mocked — ctx.tools.call(...) records the call and returns whatever you wired with MockTools.

See examples/hello-world/ for a runnable walkthrough.

Testing your app

import pytest
from linkworld_sdk.testing import TestClient
from my_app import app

@pytest.fixture
def client():
    return TestClient(app)

async def test_echo(client):
    client.tools.set_response("email_send", {"sent": True})
    await client.simulate_inbound(
        tenant_id="t1", user_id="u1", message_text="hello"
    )
    assert client.tools.calls[0][0] == "email_send"
    assert client.tools.calls[0][1]["body"] == "From u1: hello"

Decorator API

Every handler receives a Context:

ctx.tenant_id            # UUID of the tenant
ctx.user_id              # UUID of the originating user, or None
ctx.app_id               # this app's slug
ctx.event_type           # 'inbound' | 'schedule' | 'install' | 'route' | 'public_route' | …
ctx.tools                # await ctx.tools.call("tool_name", **args)
ctx.secrets              # await ctx.secrets.get("KEY")
ctx.agent                # await ctx.agent.ask("…", agent="drafter")  ← persona-aware LLM
ctx.config               # tenant install_settings answers (read-only)
ctx.platform_origin      # e.g. "https://app.linkworld.ai" (for public_callback_url)
ctx.tenant_slug          # e.g. "ocean-tec"
ctx.public_callback_url("/oauth/callback")  # full URL for THIS tenant's install
ctx.logger               # structlog-compatible logger

ctx.agent.ask — persona-aware LLM (M3.10)

Replaces direct LLM calls. Manifest declares one or more specialist agents; ctx.agent.ask routes to them with the tenant's connected LLM provider:

# manifest
agents:
  - id: drafter
    name: Post Drafter
    is_default: true
    system_prompt: |
      You write LinkedIn posts in the user's voice.
      Always plain text, ≤ 280 chars, no emojis.
  - id: classifier
    name: Comment Classifier
    system_prompt: |
      You classify comments. Output JSON only.

@app.tool("draft_post")
async def draft_post(ctx, topic: str) -> dict:
    res = await ctx.agent.ask(
        f"Write a post about {topic}.",
        agent="drafter",  # default agent if omitted
    )
    return {"draft": res.text}

@app.tool("classify_comment")
async def classify_comment(ctx, text: str) -> dict:
    res = await ctx.agent.ask(
        f"Classify: {text}",
        agent="classifier",
        response_format={
            "type": "object",
            "properties": {"sentiment": {"type": "string"}},
        },
    )
    return res.data  # parsed structured output

Per-agent tools_allowed whitelists are enforced platform-side. Costs (input + output tokens) are billed to the tenant and tagged with the app id.

@app.http_route — custom HTTP endpoints (M3.11)

Exposes raw HTTP endpoints the platform proxies to your container. Two flavours:

# Public callback — anyone on the internet can hit it. Use for OAuth
# redirects, third-party webhooks, signed downloads.
@app.http_route("/oauth/callback", methods=["GET"], public=True)
async def stripe_callback(ctx, request):
    code = request.query_params.get("code")
    state = request.query_params.get("state")
    # Verify state against ctx.secrets, exchange code for tokens, …
    return {
        "status": 200,
        "body": "<h1>Connected!</h1>",
        "headers": {"content-type": "text/html"},
    }

# Private route — requires a tenant session. Use for download URLs
# you embed in chat replies, partner-driven UIs, etc.
@app.http_route("/quotes/{quote_id}.pdf", methods=["GET"])
async def download_pdf(ctx, request, quote_id: str):
    # ctx.tools and ctx.agent are fully available here.
    pdf_bytes = await render_pdf(ctx, quote_id)
    return {
        "status": 200,
        "body": pdf_bytes,
        "headers": {"content-type": "application/pdf"},
    }

Manifest declaration is required:

http_routes:
  - path: /oauth/callback
    method: GET
    public: true
  - path: /quotes/{quote_id}.pdf
    method: GET
    public: false

Public route URL: built per-tenant via ctx.public_callback_url, typically in on_install to register with a third-party:

@app.on_install
async def on_install(ctx):
    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)

Sandbox on public routes: ctx.tools and ctx.agent raise on call — public callbacks have no tenant user, so platform tool calls have no auth principal. Read ctx.secrets for app-scoped values (OAuth state secrets, signing keys) and return a response.

Security floor enforced by the platform (you don't need to think about it, but you should know it's there):

• 1MB inbound body, 10MB outbound body — anything larger gets 413/502
• Cookie/Authorization headers NEVER forwarded to your container
• Set-Cookie/Location headers stripped from your response (no session fixation, no open-redirect via 3xx — 3xx is rewritten to 502)
• Forced Content-Security-Policy: sandbox on every response so HTML you return can't read tenant cookies even though it's served from the platform domain
• Content-Type allowlist: HTML, JSON, plain text, PDF, images. Anything else (notably application/javascript) → 502
• 30s timeout for public routes, 60s for private; per-(tenant, app) inflight cap of 10 concurrent requests — over capacity → 503
• Public M1 = GET only. POST/PUT/DELETE/PATCH on public routes return 405; mutation surface needs an HMAC-signed token (deferred)
• Path traversal, control chars, protocol-relative paths all rejected

Manifest schema

The full schema lives at packages/sdk-spec/manifest-v2.schema.yaml. Validate locally with linkworld_sdk.load_manifest("linkworld.app.yaml").

Walled-garden (M3.9)

Partner-app containers run on a locked-down Docker bridge with deny-by-default egress. Direct outbound HTTP to api.linkedin.com, api.anthropic.com, or any other public host is dropped at the firewall layer — your code can only reach the platform's MCP relay.

To talk to an external service, use a platform-mediated tool:

Note: llm_complete / llm_vision were removed in M3.10. All LLM traffic now goes through ctx.agent.ask, which respects the declared agent's system prompt and per-agent tool allowlist.

The platform resolves the tenant's stored credentials, makes the call, and bills the tenant — your app never sees an API key.

The legacy network.egress_hosts manifest field is deprecated and accepted for back-compat but ignored at runtime. New manifests should drop it; reach external services via tools instead.

License

MIT — see LICENSE.