Python SDK and CLI for xMagic, Stochastic's AI agent platform.
Status: alpha scaffold. Parts of the surface below are still stubs that raise
NotImplementedError— each is marked with the roadmap phase that implements it. See DESIGN.md for the design plan, TODO.md for current work, and CHANGELOG.md for what shipped.
Working today:
-
MCP servers — scaffold containerized (Dockerfile included) MCP servers you can register as xMagic custom tools.
-
xMagic API — chat (sync + streaming), file uploads, Drive (knowledge base), Worklists, skills packaging, with a matching async client. Request/response shapes are verified against the live API and locked with recorded-fixture tests.
-
Bring your own model —
provider:modelrefs backed by your own key.xmagic:<agent_id>,openai:<model>, andlitellm:<vendor>/<model>all work today, over oneProviderinterface.litellm:reaches the ~150 vendors LiteLLM supports, Anthropic and Google among them; their native adapters (anthropic:,google:) are reserved but unimplemented, because LiteLLM already gets you there. -
Test your code without a key —
xmagic.testing.FakeXMagicfakes the backend in process. The real client talks to it, replies are scripted, and every response has the shape recorded from the live API.
Planned:
- Local web app (Phase 5) —
xmagic serveruns the xMagic web app locally via proxy.
Requires Python 3.11–3.14.
uv pip install xmagic-sdk # core
uv pip install "xmagic-sdk[all]" # + all provider/serve/mcp extrasExtras are granular if you don't want everything — [mcp] for the server
scaffold, [serve] for the local web app, [openai] for the OpenAI provider,
and [litellm] for the long tail. There is no [anthropic] or [google]:
those adapters aren't implemented, and LiteLLM already reaches both. pip works
too if you don't use uv.
You install
xmagic-sdkbut importxmagic. The two names differ becausexmagicon PyPI belongs to an unrelated project registered in 2022, so it was never available to us.If you used this package before 0.1.0, note that releases 0.0.1–0.0.3 installed a
xmagic_sdkmodule with a completely different API. Upgrading removes it, soimport xmagic_sdkwill now raise an error explaining the change. Pinxmagic-sdk==0.0.3if you still depend on that API — see CHANGELOG.md for what moved.
From a checkout:
uv pip install -e . # core
uv pip install -e ".[all]" # + all provider/serve/mcp extrasVerify it landed:
xmagic versionIn xmagic.ai: profile → API keys. You'll also want an agent id — create an agent in Studio, and its id appears in the agent's URL.
xmagic configure # prompts for the key, hidden inputThis writes ~/.config/xmagic/config.toml with mode 600. Pass
--agent <agent_id> to set a default agent, or --api-key to skip the prompt
in a script.
Prefer environment variables? XMAGIC_API_KEY and XMAGIC_BASE_URL both work
and take precedence over the file:
export XMAGIC_API_KEY="xm-..."Full precedence is explicit arguments → environment → config file →
defaults, so you can keep a config file for everyday use and override it
per-command. XMAGIC_CONFIG_PATH relocates the file itself.
The config file looks like this, and you can edit it directly:
[xmagic]
api_key = "xm-..."
base_url = "https://api.xmagic.ai/xmagic-backend/v1"
default_agent_id = "..."
[providers.openai] # used by `-m openai:<model>`
api_key = "sk-..."Provider keys are also read from the usual OPENAI_API_KEY,
ANTHROPIC_API_KEY, and GOOGLE_API_KEY variables. Keys are only ever
written to your user config directory, never into a project folder.
xmagic workspaces # list accessible workspaces
xmagic workspaces "Workspace Name" # switch by exact name
xmagic workspaces --id <workspace_id> # switch by id
xmagic agents # list agents in current workspace contextxmagic workspaces prints each workspace's name, id, and access level.
xmagic agents config --agent <agent_id>If --agent is omitted, the CLI falls back to default_agent_id from
xmagic configure --agent .... The command fetches temporary config from
the backend, opens your editor (VISUAL, then EDITOR,
then OS default), and on save pushes the update.
xmagic agents deploy --agent <agent_id>
xmagic agents deploy --agent <agent_id> --version "Q3 rollout"
xmagic agents deploy --agent <agent_id> --phone <phone_id>
xmagic agents deploy --agent <agent_id> --no-phone # CI/non-interactive usexmagic agents deploy saves the current temporary config as a named version
and deploys it. If --version is omitted, the CLI uses the current
date/time as the version name. Without --phone or --no-phone, the command
offers optional phone and subagent association interactively. Use --no-phone
when running unattended. If VISUAL or EDITOR points to a GUI editor such as
VS Code, include its wait flag (for example, code --wait) when editing YAML.
xmagic chat --agent <agent_id> "Summarize our Q3 goals" # one-shot
xmagic chat --agent <agent_id> # interactive sessionResponses stream by default; --no-stream waits for the full reply instead.
If you set default_agent_id during configure, drop the --agent flag.
When the agent thinks out loud, its reasoning is printed dimmed, above the
answer.
Attach files with -f (repeatable), and pick the chat's UI context with
--chat-type:
xmagic chat --agent <agent_id> -f notes.md -f data.csv "What changed?"
xmagic chat --agent <agent_id> --chat-type playground "Try something"An interactive session reuses a single chat, so the agent keeps its context across turns.
For scripts, every command that produces data takes --json: stdout is then
one JSON document and nothing else, errors go to stderr, and the exit code is
the verdict, so the output pipes straight into jq:
xmagic chat --agent <agent_id> --json "Summarize our Q3 goals" | jq -r .text
xmagic agents --json | jq -r '.[].id'
xmagic worklists trigger <task_id> --json | jq .statusTo get the answer in a shape a script can rely on, hand chat a JSON Schema
file with --schema. The reply is validated against it before anything is
printed, and under --json the instance is on parsed:
cat > weather.json <<'EOF'
{"type": "object",
"properties": {"city": {"type": "string"}, "temp_c": {"type": "number"}},
"required": ["city", "temp_c"]}
EOF
xmagic chat -m openai:gpt-5 --schema weather.json --json "Weather in Osaka?" | jq .parsed.temp_cA reply that does not match fails with a non-zero exit and pydantic's reasons on
stderr, never a half-valid document on stdout. --schema works with openai:
and litellm: refs; an xMagic agent's output shape is part of its dashboard
configuration, so --agent rejects it. The file may use object, array, scalar,
enum, anyOf, and nullable types, required, description, default, and
the numeric and length constraints; anything else ($ref, format, oneOf,
...) is refused by name rather than silently dropped.
List one page of background tasks, inspect a task and its latest result, or control its execution:
xmagic worklists --agent <agent_id>
xmagic worklists get <task_id> --agent <agent_id>
xmagic worklists cancel <task_id> --agent <agent_id>
xmagic worklists delete <task_id> --agent <agent_id> --yes
xmagic worklists trigger <task_id> --agent <agent_id>
xmagic worklists rerun <task_id> --agent <agent_id>
xmagic worklists review [task_id] --agent <agent_id>xmagic worklists create opens a pre-filled YAML template, while edit and
schedules edit open the current editable fields in the configured editor
(VISUAL, then EDITOR, then the platform default). Save the file to submit
the changes. List pagination is deliberately single-page: use --skip and
--limit (1–200) when fetching another page; the CLI reports the page size and
total count instead of silently making additional requests.
worklists review displays the latest result for each task in needs_review,
one at a time. Enter a message to send guidance to the agent in the task's
existing chat thread. Press Enter without a message to complete the task
without another agent action, or type /skip to leave it in needs_review for
later. Pass a task ID to review one task.
Recurring schedules can also be inspected and controlled with
xmagic worklists schedules get|edit|pause|resume|delete.
A task's inputs are storage paths (input_s3_file_paths). To use local files,
pass them with --input on create or edit, or list them under
input_files in the YAML; on save each is uploaded into a Drive folder and its
path appended to the task's inputs:
xmagic worklists create --agent <agent_id> --input notes.md --input data.csv
xmagic worklists edit <task_id> --agent <agent_id> --input more.pdf --folder <folder_id>The files land in a Drive folder named worklist-inputs (created the first
time) unless --folder gives another folder's id, so they stay visible and deletable in
Drive. From Python the same step is client.worklists.upload_inputs(folder_id, paths), which returns the paths to put in input_s3_file_paths.
Drive is the knowledge base your agents retrieve from. Every route the client implements is on the command line:
xmagic drive ls # folders
xmagic drive ls <folder_id> # files in one folder
xmagic drive ls -R # every top-level folder and its files
xmagic drive mkdir "Q3 reports"
xmagic drive info <folder_id> # one folder, with counts
xmagic drive rename <folder_id> "Q3 reports (final)"
xmagic drive upload <folder_id> notes.md # indexed automatically
xmagic drive download <folder_id> <file_id> [<file_id>...] --extract ./out
xmagic drive rm <folder_id> <file_id> # delete files
xmagic drive rm <folder_id> --yes # delete the folder and everything in itxMagic exports files as one ZIP archive, even for a single file. download
writes it as <folder_id>.zip by default, --output names it, and
--extract DIR unpacks it instead. rm with no file ids deletes the folder,
so it asks first unless --yes. Every command takes --json.
from xmagic import XMagicClient
client = XMagicClient() # reads env/config; or XMagicClient(api_key="xm-...")
chat = client.chats.create("<agent_id>", title="demo")
# Streaming. The stream is closable: `with` releases the connection the moment
# the block ends, and close() mid-loop cancels the query outright.
with client.chats.stream("<agent_id>", chat.id, "Explain xMagic skills") as events:
for event in events:
if event.type == "response":
print(event.text, end="")
if "enough" in event.text:
events.close() # cancel now; nothing else arrives
# Blocking
resp = client.chats.query("<agent_id>", chat.id, "One-sentence summary?")
# Non-interactive worklist review. No message completes the task without
# another agent action; a message continues the existing worklist run chat.
review = client.worklists.review("<agent_id>", "<task_id>", message="Approved")
print(review.action, review.task.id)XMagicClient is a context manager, so with XMagicClient() as client: closes
the underlying HTTP connection for you. It retries 429 and 5xx with
jittered exponential backoff, honoring Retry-After.
A stream left mid-loop without with or close() is released only when the
garbage collector reaches it — soon on CPython for the sync client, but for the
async client only when the event loop's finalizer runs, which after the loop is
gone is never. Provider.stream() returns the same kind of closable stream.
Timeouts come in two flavours, because streams need a looser bound than unary
calls — timeout (default 60s) bounds a whole request, while stream_timeout
(default 300s, None waits forever) bounds the gap between two stream events,
so an agent that thinks for a while is not mistaken for a dead connection:
client = XMagicClient(timeout=30.0, stream_timeout=600.0, max_retries=5)When a call needs inspecting, the client logs under the xmagic logger and is
silent until you ask — request and response lines at DEBUG, retries at
INFO, and never headers or bodies, so the API key cannot end up in a log
file. xmagic -v <command> turns it on for the CLI; from Python:
import logging
logging.basicConfig(level=logging.DEBUG) # or just logging.getLogger("xmagic")Every request also carries a User-Agent naming the SDK, Python, and httpx
versions, so a problem on the server side can be tied to the version that hit
it.
Everything that can go wrong raises a subclass of XMagicError, so one except
contains the SDK — including connection failures, which are wrapped rather than
leaked as httpx exceptions:
from xmagic import APIConnectionError, RateLimitError, XMagicAPIError, XMagicError
try:
resp = client.chats.query("<agent_id>", chat.id, "Hello!")
except RateLimitError:
... # 429, after the retries were exhausted
except APIConnectionError:
... # never got a response at all; APITimeoutError is a subclass
except XMagicAPIError as e:
print(e.status_code, e.request_id) # also .error_code, .response, .headers, .body
except XMagicError:
... # everything else, e.g. ConfigurationErrorThe status-specific classes are BadRequestError (400), AuthenticationError
(401), PermissionDeniedError (403), NotFoundError (404), RateLimitError
(429), and ServerError (any 5xx).
The package ships a py.typed marker, so mypy and pyright see its annotations
rather than Any.
AsyncXMagicClient mirrors it 1:1 — same resources, same arguments, same
returns. Await each call, and iterate stream with async for:
from xmagic import AsyncXMagicClient
async with AsyncXMagicClient() as client:
chat = await client.chats.create("<agent_id>", title="demo")
async with client.chats.stream("<agent_id>", chat.id, "Explain xMagic skills") as events:
async for event in events:
if event.type == "response":
print(event.text, end="")
review = await client.worklists.review("<agent_id>", "<task_id>")
print(review.action, review.task.id)xmagic mcp init my-tool # scaffold: Dockerfile, compose, MCP server
cd my-tool && docker compose up --buildThat serves MCP over streamable HTTP at http://localhost:8000/mcp. xMagic
needs a public HTTPS URL to reach it, so for development expose it with a
tunnel:
cloudflared tunnel --url http://localhost:8000Before going anywhere near a tunnel, exercise it locally — tools list and
tools call speak MCP straight to the server:
xmagic tools list --url http://localhost:8000/mcp --api-key "$TOOL_API_KEY"
xmagic tools call ping --url http://localhost:8000/mcp --api-key "$TOOL_API_KEY" -a message=hiThat skips the whole tunnel → register → open a chat → hope-the-agent-calls-it
loop, which otherwise makes a broken tool and a tool the agent declined to use
look identical. Pass --json for scriptable output; call exits non-zero when
the tool reports an error, so it works in CI. Repeat -a key=value for multiple
arguments, or pass --json-args '{"k": "v"}'.
Then register the resulting public https://.../mcp URL in the dashboard under
Custom tools → Create tool. xmagic tools register --name ... --url ...
prints the full checklist. Set TOOL_API_KEY in your .env to require a
shared secret — the generated server rejects unauthenticated calls with 401.
xmagic skills new my-skill # scaffold SKILL.md
xmagic skills validate my-skill # check frontmatter and layout
xmagic skills pack my-skill # -> my-skill.zip, ready to uploadUpload the zip in the dashboard under Skills.
chat takes a provider:model ref backed by your own key. OpenAI is
implemented:
export OPENAI_API_KEY="sk-..."
xmagic chat -m openai:gpt-5 "Hello!"Or from Python, through the same Provider interface the xMagic path uses:
from xmagic.providers import ChatMessage, get_provider
provider = get_provider("openai:gpt-5") # reads OPENAI_API_KEY
for chunk in provider.stream([ChatMessage(role="user", content="Hello!")], model="gpt-5"):
print(chunk.text, end="")Two differences from the xMagic path are worth knowing. model is a real model
name here, not an agent id — OpenAI picks per call, so there's no agent and no
server-side session, and multi-turn context is whatever you pass in messages.
And -f/--file is xMagic-only, since uploads are an xMagic feature.
Errors arrive as this SDK's own types, so you catch XMagicError regardless of
which vendor failed — a 429 from OpenAI raises the same RateLimitError a 429
from xMagic would.
Tools are a typed surface, not a **params passthrough — and the SDK runs
the loop for you:
from xmagic.providers import ChatMessage, ToolDef, get_provider
def get_weather(city: str) -> str:
"""Look up the weather in a city."""
return "sunny, 24C"
provider = get_provider("openai:gpt-5")
messages = [ChatMessage(role="user", content="What's the weather in Osaka?")]
tool = ToolDef.from_callable(get_weather) # schema from the signature
run = provider.run_tools(messages, model="gpt-5", tools=[tool])
print(run.completion.text) # "Sunny and 24C in Osaka."run_tools calls the model, runs each tool it asks for, feeds the results
back, and repeats until the model answers in text or max_turns (default 10)
model calls have been made — at which point it raises ToolTurnsExceeded
carrying a resendable history rather than returning what would look like an
answer. A raising tool, an unknown name, or arguments that fail validation go
back to the model as the tool result so it can correct itself; pass
raise_errors=True to get the exception instead. run.messages is the full
history, ready for the next user turn, and run.usage sums token counts
across every turn. From async code, await provider.arun_tools(...) — it
awaits async def tools and keeps the event loop free during model calls.
That is the whole surface: tools and a cap, no planners, memory, or retry policy — the loop is a primitive, not a framework (DESIGN.md §13.8). If you need different mechanics, the pieces are public and the loop is thirty lines:
completion = provider.complete(messages, model="gpt-5", tools=[tool])
for call in completion.tool_calls:
result = get_weather(**call.arguments) # arguments arrive parsed, not as JSON text
messages += [
ChatMessage(role="assistant", content=None, tool_calls=completion.tool_calls),
ChatMessage(role="tool", tool_call_id=call.id, content=result),
]
print(provider.complete(messages, model="gpt-5", tools=[tool]).text)Works the same through litellm: refs, since LiteLLM normalizes every vendor
onto the shape this maps.
Streaming works too. Arguments arrive as JSON fragments spread across deltas, so
a completed call lands on the terminal chunk — the same place usage and
the response id do — rather than mid-stream, where a half-built call could
not be told apart from a finished one:
for chunk in provider.stream(messages, model="gpt-5", tools=[tool]):
print(chunk.text, end="")
if chunk.done:
for call in chunk.tool_calls:
print(get_weather(**call.arguments))Every completion carries the provider's id for it: xMagic's message_id,
which chats.get_message takes, or OpenAI's chatcmpl-.... On a stream it is
on the terminal chunk. xmagic chat --json reports it as id.
One limit worth knowing: xmagic: refs reject tools= — an xMagic agent's
tools are registered in the dashboard and attached to the agent, which is a
different capability, so capabilities()["tools"] reports False there.
Structured output is a pydantic model in, an instance out:
from pydantic import BaseModel
class Weather(BaseModel):
city: str
temp_c: float
completion = provider.complete(messages, model="gpt-5", response_format=Weather)
print(completion.parsed.temp_c) # a Weather, validated -- or an exception, never NoneThe model's schema is sent as the vendor's json_schema response format, with
strict mode claimed for a flat, fully-required model (the same rule tools use).
A reply that does not validate raises, and so does a safety refusal, in the
vendor's words. On stream() the JSON arrives as text and the instance rides
the terminal chunk as parsed, where usage and tool_calls already are.
xmagic: refs reject response_format= too: an agent's output shape is part
of its configuration, not a per-call parameter. On the command line the same
thing is xmagic chat --schema weather.json, which takes the JSON Schema file
and builds the model for you (section 6).
Which refs exist is discoverable rather than guesswork:
xmagic models providers # 85 providers LiteLLM can reach
xmagic models list -p groq # their models, with tools/vision and prices
xmagic models list -s llama-3.3-70b --json # scriptableThat catalogue is LiteLLM's. xMagic publishes no model list — a xmagic: ref
names an agent, not a model — so xmagic models covers every provider except
that one, and says so in its output.
anthropic: and google: are not implemented — use litellm:anthropic/<model>
or litellm:gemini/<model> instead. Two things differ on the LiteLLM path:
credentials come from whatever environment variable the vendor uses
(ANTHROPIC_API_KEY, GROQ_API_KEY, and so on) rather than from xMagic config,
and parameters are validated against the model before the request is sent, so an
unsupported one fails locally instead of at the vendor.
Runnable scripts for each of the flows above live in
examples/ — chat, streaming, files and Drive, the MCP scaffold
walkthrough, skills packaging, bring-your-own-model, and tool calling. The last
four need no xMagic key.
xmagic --help lists every command, and each subcommand takes --help too.
See DESIGN.md for how the pieces fit together.
No API key configured — run xmagic configure, or export
XMAGIC_API_KEY. Check what's actually being picked up with
xmagic configure --help and remember env vars override the config file.
A command prints ... lands in Phase N (see DESIGN.md) — that feature is
scaffolded but not implemented yet. See the status note at the top of this
README.
ModuleNotFoundError: No module named 'xmagic' — the import name is
xmagic, but the package to install is xmagic-sdk. See the note under
Install.
ImportError mentioning xmagic_sdk — you upgraded from 0.0.x, where the
module was called xmagic_sdk. It is xmagic from 0.1.0 on, with a different
API; the error text names what changed.
401 from your MCP server — the generated server requires TOOL_API_KEY
when set. Send it as either x-api-key or Authorization: Bearer <key>.
xMagic can't reach your MCP server — it must be public HTTPS. localhost
won't work; use a tunnel for development.
xmagic.testing.FakeXMagic stands in for the backend so tests of code built
on this SDK need no key and no network. The real client runs; only the
transport is swapped.
from xmagic.testing import FakeXMagic
fake = FakeXMagic()
fake.agent("agent-1").replies("Paris") # str, or a (query) -> str callable
client = fake.client() # XMagicClient bound to the fake
chat = client.chats.create("agent-1")
assert client.chats.query("agent-1", chat.id, "Capital of France?").text == "Paris"
assert fake.calls[-1].json == {"query": "Capital of France?", "is_stream": False}Chats, streaming, uploads, and Drive are faked, with responses rendered from
the fixtures recorded from the live API (also exposed as
xmagic.testing.load_fixture for mocking a route yourself). Anything else
fails with a BadRequestError naming the route. fake.async_client() mirrors
it, XMagicProvider(client=fake.client()) puts it behind the xmagic:
adapter, and fake.fail_next(429) exercises your error handling through the
real retry loop. See examples/08_offline_tests.py.
uv sync --all-extras
uv run pytest
uv run ruff check .
uv run ruff format --check .
uv run mypyCI runs all five across Python 3.11–3.14, and gates on formatting and types as
well as linting — run uv run ruff format . before pushing.
mypy runs in strict mode over src/ and tests/. The package ships a py.typed marker,
so its annotations are what downstream type checkers believe about it; a wrong
one is worse for a consumer than none at all. tests/ is checked the same way.
Commit messages follow Conventional Commits
(feat:, fix:, docs:, test:, chore:, ...; optional scope, e.g. feat(mcp): ...).
Contributions are welcome — see CONTRIBUTING.md for setup and the PR workflow, and ISSUES.md for filing bugs, feature requests, and security reports. All participants are expected to follow our Code of Conduct.
Apache-2.0 © Stochastic
