Shadow-OS Python SDK

The official Python client for Shadow-OS — build, operate and talk to AI agents.

pip install shadow-os

from shadow_os import ShadowOS

with ShadowOS() as client:          # reads $SHADOW_OS_API_KEY
    print(client.chat("Summarise the notes I uploaded yesterday"))

• Complete — the assistant, conversations, memory, files & workspace, documents & search, the full agent platform, and the automation the agent creates for you.
• Bring your own model — run everything on your provider and your key (OpenAI, Anthropic, Google, or any OpenAI-compatible service). Your key is never stored.
• Typed — every response is a dataclass, the package ships py.typed, and unknown server fields stay reachable via .raw.
• Sync and async — ShadowOS and AsyncShadowOS share one definition per endpoint, so they cannot drift.
• Resilient — timeouts, exponential backoff with jitter, Retry-After support, and automatic retries for cold starts and 5xx.

Authentication

Two credentials, for two different jobs.

client = ShadowOS("sk-shadow-…")            # or set SHADOW_OS_API_KEY
agent  = AgentClient("lnk_…")               # or set SHADOW_OS_AGENT_TOKEN

Create an account key at /developers.

Your account assistant

reply = client.chat("What did we agree with Dana?", session_id="conv-42")
print(reply.answer, reply.usage)

session_id is an isolation boundary, not just a label: memory, uploaded files and search are scoped to it, so one end-user can never see another's. Pass scope="account" to opt into the shared account space (which also has access to connected integrations such as Gmail).

Documents & semantic search

client.documents.upload("handbook.pdf", session_id="conv-42")
for doc in client.documents.list(session_id="conv-42"):
    print(doc.name)

for hit in client.documents.search("refund policy", top_k=5, session_id="conv-42"):
    print(hit.score, hit.source, hit.text[:120])

Quota

u = client.usage()
print(f"{u.used}/{u.quota} used — {u.remaining} left")

Bring your own model

Run the whole platform — tools, memory, knowledge, agents — on your provider and your key.

from shadow_os import ShadowOS, Model

client = ShadowOS()

client.chat("Analyse this contract", model=Model.google("gemini-2.5-pro", api_key="AIza..."))
client.chat("Draft the reply",       model=Model.openai("gpt-4o", api_key="sk-..."))
client.chat("Think hard",            model=Model.anthropic("claude-sonnet-4-5",
                                                           api_key="sk-ant-..."), effort="high")

print(client.providers().available)     # what this deployment can actually serve

client.chat("Summarise", model=Model.compatible(
    "llama-3.3-70b", api_key="gsk-...", base_url="https://api.groq.com/openai/v1"))

Your key is never stored. It is sent with the request that uses it, used for that single turn, and discarded — never written to a database, an agent's config, a log line or a backup. That is why it is passed per call rather than configured once: there is nothing to configure, by design. Model also redacts it from repr, so it cannot leak into a traceback.

It works on agents too, and on AgentClient:

client.agent(agent_id).talk("Draft a reply to Dana")
AgentClient("lnk_...").send("Hi", model=Model.google("gemini-2.5-pro", api_key="AIza..."))

effort is "low" | "medium" | "high" and maps onto whatever the provider calls reasoning depth. It works with or without a model — on its own it sets the depth of the platform model.

Local tools — the agent works on your computer

The agent thinks in the cloud — model, memory, the whole tool loop. With a few lines of Python it can also act on the computer running your code: your own functions, a real browser with your logins, a folder, programs you allow, any MCP server. No server to run, no open port, no tunnel, nothing to deploy — the SDK opens one outbound connection, and nothing on the machine is reachable except what you list.

Your functions are the tools

from shadow_os import ShadowOS

def stock(item: str) -> int:
    """How many units of an item we have."""    # type hints + docstring ARE the tool
    return inventory[item]

client = ShadowOS()
reply = client.chat("Do we have enough flour for Friday?", tools=[stock])
print(reply)               # the answer
print(reply.local_calls)   # what ran here: [{"tool": "stock", "ok": True, "ms": 4}]

Async functions, local.expose(module_or_object) and local.tool(fn, timeout=..., audience=...) all work. Return anything JSON can carry (dataclasses, dates, paths, pydantic models are converted); an exception reaches the agent as an error it can explain — the traceback stays on your machine.

An agent in code that works toward a goal

from shadow_os.local import Browser          # pip install "shadow-os[browser]"

with client.session(tools=[Browser(visible=True)],
                    instructions="You are a careful player. Never spend real money.") as s:
    s.ask("Open https://mykabinet.co.il/#/battle and log me in")        # one turn
    result = s.run("Win 3 battles and tell me the trophy count", max_turns=20)
    print(result.success, result.result, result.turns, result.interruptions)

run() keeps the agent working turn after turn until it calls finish(result, success) (a tool the SDK adds) or reaches max_turns. If a turn is cut off — e.g. a deploy restarts the server — it waits for this computer to reconnect and continues, telling the agent to check the current state first so nothing is done twice. on_call=fn on the session sees every local call as it happens.

A real browser

Browser(visible=True) drives Chromium on this computer. The agent reads the page as text with numbered elements and clicks or types by number — browser_click_many clicks several in one move — and the click still lands when a single-page app redraws itself, changes the button's text, or covers it with an invisible layer. Every action hands back the page's current numbered elements, so the agent never acts on stale numbers. Logins stay in a saved profile (~/.shadow_os/browser/default), so you sign in once; browser_wait_for_user lets the agent pause while you log in or approve something. Browser(attach="http://localhost:9222") drives your own Chrome started with --remote-debugging-port=9222. Long browser tasks go best on a strong model — pass model=.

The brain on your computer too — a local model

from shadow_os import local

with client.session(tools=[local.Browser(visible=True)], model=local.Ollama("qwen3")) as s:
    s.run("Win 3 battles")

The platform keeps the agent — prompts, memory, the tool loop, channels — and every completion comes to this computer over the same connection and goes to your local OpenAI-compatible server: local.Ollama(model), local.LMStudio(model), or local.LocalModel(model, base_url="http://gpu-box:8000/v1") for vLLM, llama.cpp and the rest. Your GPU does the thinking and inference costs nothing; the conversation still passes through the platform. Works on chat, session, run and talk, and only with your account key — a share link can never point turns at your machine. Pick a model that is good at tool use; small ones often are not.

Ready packs, and files back

from shadow_os import local

tools = [
    local.Folder("~/Shop", write=False),       # list, read, search — nothing outside this folder
    local.Shell(allow=["git", "ffmpeg"]),      # only these programs, never a shell
    local.Python(),                            # code in a fresh isolated process (opt-in)
    local.MCP("npx -y @modelcontextprotocol/server-filesystem ~/Shop"),   # any stdio MCP server's tools
]

def sales_report() -> local.File:              # a FILE comes back to the agent's workspace,
    return local.File("sales.xlsx")            # where it can read it or send it to someone

Always-on: your computer serves your agent

def stock(item: str) -> int:
    """How many units of an item we have."""
    return excel[item]

def refund(order_id: str) -> str:
    """Refund an order."""
    ...

client.agent("ag_123").serve(tools=[
    local.tool(stock, audience="customers"),   # anyone talking to the agent (WhatsApp, web, API)
    refund,                                    # default "owner": only you
], reload=True)                                # edit the file -> it restarts by itself

Each tool says who may trigger it: owner (default — only you), staff (anyone with a manager link) or customers. A customer can never reach an owner tool, even by naming it. Several computers can serve one agent; a turn that already acted on a computer is never replayed after a server restart. agent.machines() — and the console's Your computers card — show what is connected right now.

Events — your code wakes the agent

runner = client.agent("ag_123").serve(tools=[stock], block=False)
res = runner.emit("A new invoice for Dana is ready - send it to her", data={"file": "inv-17.pdf"})
print(res["answer"], res["actions"])

An event runs as a turn with your authority, so the agent can act on it — including messaging a customer.

What you can rely on

• Every local call prints on your console as it happens (quiet=True on a LocalRunner silences it).
• Nothing runs twice. A turn with local tools is never retried. If it is cut off after something ran you get TurnInterrupted with .local_calls (a goal run carries on by itself).
• If the connection drops — Wi-Fi, sleep, a server restart — the SDK reconnects, and a turn waits for it.
• Limits: 120 s per call by default (up to 30 min), results up to 20,000 characters, files up to 8 MB, 120 local calls and 30 events a minute per computer.

Conversations & memory

A conversation loads its own messages, so there is nothing to work out:

for conv in client.conversations.list():
    print(conv.session_id, conv.title)
    for msg in conv.messages():
        print(" ", msg.role, msg.content)

# or, if you already hold the id
for msg in client.conversations.history("thread-id"):
    print(msg.role, msg.content)

client.conversations.rename("thread-id", "Contract review")
print(client.memory.summary())          # what it has learned about you, across everything

The same shape works for an agent's members:

with AgentClient("lnk_...") as agent:
    for conv in agent.conversations(member_key="user-42"):
        for msg in conv.messages():
            print(msg.role, msg.content)

Files & workspace

Two different things, and the difference matters:

client.workspace.index("handbook.pdf")   # extract text → findable by semantic search
client.workspace.add("analysis.py")      # keep the RAW file → the agent can edit and RUN it
client.workspace.list()
client.workspace.download("report.xlsx") # something the agent produced

Automation

You don't build automation through this API — you ask for it. The agent creates and runs recurring work itself, from an ordinary message. These methods exist to see and control what it already made, which is what you need for a dashboard.

client.chat("Every weekday at 08:00, email me a summary of my unread mail")

for t in client.tasks.list():        # scheduled work
    print(t.description, t.target_time, t.status)
client.tasks.reschedule(t.id, "2026-10-01T08:00:00")

for w in client.watches.list():      # monitors it created
    print(w.title, w.status)
client.watches.pause(w.id)

for r in client.recipes.list():      # reusable workflows it saved
    for run in client.recipes.runs(r.id):
        print(run.status, run.started_at)

Creating them directly is possible (client.recipes.create(), client.watches.create()) but it is the harder path: the agent writes better steps from a sentence than you will from a schema.

Notifications & integrations

client.notifications.set_prefs(scheduled_results=True, owner_alerts=False)
client.integrations.google()        # Gmail + Calendar connected?
client.integrations.whatsapp()

Building an agent

from shadow_os import ShadowOS, AgentConfig

client = ShadowOS()

created = client.agents.create(
    name="Nona Pizza",
    template="pizzeria",                       # or persona="..." for full control
    fields={"business_name": "Nona", "hours": "Sun–Thu 11:00–23:00"},
    config=AgentConfig(memory_mode="single", tone="warm and brief"),
)
print(created.id, created.manager_token)       # the token is shown ONCE — store it now

Or start from a draft: the platform reads your website, a document or a few words and proposes the name, instructions, welcome line, hours, prices and FAQ. Nothing is created until you call create.

draft = client.agents.draft(url="https://nona.example", notes="We also cater events")

Browse what's available before you build:

for t in client.templates():
    print(t.id, t.name, [f["id"] for f in t.fields])

print(client.tool_catalog().groups)            # optional tool bundles you can enable per agent

Teaching it

agent = client.agent(created.id)

agent.knowledge.add_file("menu.pdf", description="Full menu with prices")
agent.knowledge.add_file("storefront.jpg")     # images are understood AND sendable to customers
agent.knowledge.add_url("https://nona.example/about")
agent.knowledge.add_text("We deliver within 4km. Closed on holidays.", title="policies")

agent.knowledge.add_sendable("invoice-template.pdf")   # deliverable, but NOT searchable knowledge

Short facts go in the fact book — shown to the agent on every turn, and kept current: a newer version of a fact replaces the old one instead of piling up next to it.

agent.facts.add("Delivery is free above 120 NIS")
agent.facts.add("Delivery is free above 150 NIS")   # replaces the previous fact
for f in agent.facts.list():
    print(f["id"], f["text"])

Tools the agent built for itself in the Manage chat wait for your approval:

for t in agent.tools.list():
    print(t["name"], t["status"])
agent.tools.approve("weather_now")

Sharing it

link = agent.share_link()                      # persistent, created once, stable forever
print(link.web, link.whatsapp)

staff = agent.access.create_share_link(role="manager", label="Front desk")
api   = agent.access.create_token(role="member", label="Website widget")

Operating it

for c in agent.customers():
    print(c.display_name, "·", c.profile)

for e in agent.escalations():
    agent.answer_escalation(e.id, "Yes — we're open until 23:00 on Sunday.")

print(agent.analytics().raw)
print(agent.appointments())
print(agent.delivery_status())

Campaigns follow up quiet leads and invite customers back — each message written per person:

print(agent.campaigns.preview("quiet_leads"))       # who is due, and a sample message; sends nothing
agent.campaigns.run("quiet_leads")

Talk to it as its manager — it sees your customer roster and acts with your authority:

print(agent.talk("Message Dana that her order is ready"))

Talking to an agent (what you embed)

from shadow_os import AgentClient

with AgentClient("lnk_…") as agent:
    info = agent.info()
    print(info["agent_name"], info["welcome"])

    reply = agent.send("Are you open on Sunday?", member_key="user-7f3c")
    print(reply.answer, reply.files)

member_key is the identity of the person you are speaking for. Give each end-user a stable, unguessable id and each gets private, durable memory and history inside the agent — they can never see each other.

dana = agent.for_member("user-7f3c", member_name="Dana")   # shares the connection pool
for conv in dana.conversations():
    print(conv.session_id, conv.title)
for msg in dana.history(session_id="main"):
    print(msg.role, msg.content)

Async

Identical surface, awaited:

from shadow_os import AsyncShadowOS

async with AsyncShadowOS() as client:
    reply = await client.chat("hello")
    agents = await client.agents.list()
    stats  = await client.agent(agents[0].id).analytics()

Errors

from shadow_os import QuotaExceeded, RateLimited, ShadowOSError

try:
    client.chat("hello")
except QuotaExceeded as e:
    print("out of quota:", e.usage)
except RateLimited as e:
    print("slow down, retry in", e.retry_after)
except ShadowOSError as e:
    print(e.status, e.code, e.message, e.request_id)

Every error carries status, code, message and — when the server sent one — request_id. Quote that id in a support request and the exact call can be found in the logs.

Configuration

client = ShadowOS(
    api_key="sk-shadow-…",       # or $SHADOW_OS_API_KEY
    base_url="https://…",        # or $SHADOW_OS_BASE_URL — for self-hosting or staging
    timeout=120.0,               # per request; generous on purpose (the service can cold-start)
    max_retries=3,               # timeouts, connection errors, 429 and 5xx
)

Bring your own httpx client (proxies, custom TLS, shared pools):

import httpx
client = ShadowOS(http_client=httpx.Client(proxies="http://…", timeout=60))

Retries use exponential backoff with full jitter. Retry-After always wins. 4xx responses other than 429 are never retried — they will not become a 200.

Development

pip install -e ".[dev]"
pytest

MIT licensed.