m8tes Python SDK

Run agents from Python with 190+ integrations, memory, streaming, and per-user isolation.

Install

pip install -U "m8tes>=4.8"

Quick start

1. Create an account and save your API key.
2. Open Account → Model connections, connect xAI, and finish provider sign-in. Wait for Connected.
3. Set M8TES_API_KEY, then stream a reply:

from m8tes import M8tes

client = M8tes()
# Personal development only: this disables strict user_id checks account-wide.
# This persists. For customer-facing apps, keep strict mode on and pass user_id.
client.settings.update(require_end_user_id=False)

for text in client.runs.stream_text(
    message="Draft a warm reply to a customer asking to cancel.",
    model="grok-4.6",
    raise_on_error=True,
):
    print(text, end="", flush=True)

For production, keep strict mode on and pass user_id for each customer. These runs use prepaid funds; new balances start at $0. Top up before running.

Auth & usage

Rotate your API key with POST /api/v2/token. That endpoint returns a new API key and invalidates the previous one.

Check current plan, run usage, and cost limits with client.billing.usage() (or client.auth.get_usage()). A billable run is one execution that completes with output — manual, scheduled, webhook, email, reply, or retry. Self-meter spend and control overage:

usage = client.billing.usage()
print(usage.plan, usage.runs_used, usage.runs_limit, usage.overage_used_cents)

# Browse Hobby, Individual ($20 with your model subscription), and team plans
for plan in client.billing.plans(include_free=True):
    print(plan.slug, plan.display_name, plan.included_runs, plan.monthly_price_cents)

# Platform-inference team plans expose plan.overage_available=true and can opt in
# to usage overage with a monthly spend cap.
client.billing.set_overage(enabled=True, monthly_cap_cents=5000)  # $50 cap

New accounts start unfunded. Connect a model subscription to activate the $0 Hobby plan immediately (150 runs every 30 days), choose Individual for $20/month and 1,000 runs using that subscription, or choose a team plan starting at $1,000/month with inference included.

Enable an @notifications.m8tes.ai inbox per agent with email_inbox=True on client.agents.create(...) or call client.agents.enable_email_inbox(agent_id) later.

Need iMessage-triggered runs? Configure BlueBubbles on your account, then set inbound_imessage_enabled=True and imessage_chat_guid="..." on client.agents.create(...) or client.agents.update(...). Use a dedicated 1:1 chat unless you intentionally want everyone in that thread to trigger the agent and receive its replies.

Inspect account request history with client.audit_logs.list(...):

page = client.audit_logs.list(method="POST", resource_type="run", limit=10)
for log in page.data:
    print(log.created_at, log.method, log.path, log.status_code)

Use cases

Revenue reporting. Pull MRR from Stripe, update the tracking sheet, post weekly delta to Slack. No more manual Monday reporting.

Support triage. Classify inbound tickets, draft replies, escalate blockers. Runs 24/7 on a schedule.

Ad spend monitoring. Check Google Ads weekly, pause low-converting campaigns, alert the team.

Customer-facing agents. Give each user their own agent with isolated memory, tools, and permissions. Multi-tenant without custom plumbing.

vs. eve, LangChain, CrewAI, and other frameworks

eve, LangChain, CrewAI, and the OpenAI Agents SDK are agent frameworks. They help you build one production agent — but you still write, deploy, and operate the agent application, and execution, OAuth, scheduling, memory, approval flows, and tenant isolation are all yours to build and host.

m8tes is not a framework. It's the layer above one: a framework gives you a codebase to build an agent; m8tes gives you an API to give every customer one. The Python SDK is the client on top.

Models

Pick the model per agent or per run via model=. List what's available (with prices) instead of hardcoding:

for m in client.models.list().data:
    print(m.id, m.provider, m.pricing.input_per_mtok, "→", m.pricing.output_per_mtok, "/Mtok")

bot = client.agents.create(name="Ops", model="sonnet")  # or per run: runs.create(..., model="opus")

Today that's the Claude models sonnet, opus, and fable (Fable 5.1 — curated picker; ~2x Opus cost); OpenAI gpt-6-astra (frontier, curated picker) / gpt-5.5 / gpt-5.6-sol / gpt-5.6-terra / gpt-5.6-luna; Google gemini-3.8-flash / gemini-3.7-flash / gemini-3.6-flash; xAI grok-4.6 (platform default, curated) / grok-4.5; Meta muse-spark-1.3 (no ZDR host — API-only); and open-source glm-5.2 / minimax-m3 / deepseek-v4-pro-0813 / deepseek-v4-pro / deepseek-v4-flash-0731 (lowest cost) / deepseek-v3-2 / kimi-k2-7-code / kimi-k3 / qwen3.8-max. Zero-data-retention support is per model and changes over time, so never assume it from the model name: read zdr_supported on GET /api/v2/models, or filter with ?zdr=true, before sending customer data — claude-fable-5-1 in the curated picker and muse-spark-1.3 on the API are the documented exceptions with no ZDR host. models.list() is the live source of truth; omit model to use the default.

Own provider subscription

Connect a personal Claude, Codex, Grok, or Gemini plan under Account → Model connections so matching account-scoped runs bill that provider instead of prepaid credits. The credential never attaches to a run created with user_id.

# Codex / Grok: device code, then poll until connected
auth = client.model_connections.authorize("openai")  # or "xai"
print(auth.authorization_url, auth.user_code)
status = client.model_connections.authorization_status("openai", auth.state)

# Gemini: paste the code from Google (no device code)
auth = client.model_connections.authorize("gemini")
print(auth.authorization_url)  # user_code is None
client.model_connections.complete_authorization("gemini", auth.state, code="...")

print([c.provider for c in client.model_connections.list().data])

Runs

Streaming (default)

for event in client.runs.create(
    message="pull MRR from Stripe, compare to last month, post the delta to #revenue",
    tools=["stripe", "slack"],
):
    match event.type:
        case "text-delta":
            print(event.delta, end="")
        case "tool-call-start":
            print(f"\n  {event.tool_name}")
        case "tool-result-end":
            print(f"  > {event.result[:100]}")
        case "done":
            print(f"\n  {event.stop_reason}")

Non-streaming

run = client.runs.create(message="generate quarterly report", stream=False)
result = client.runs.poll(run.id)  # blocks until complete
print(result.output)

# or use the convenience wrapper
result = client.runs.create_and_wait(message="generate quarterly report")

Context manager

with client.runs.create(message="summarize inbox") as stream:
    for event in stream:
        print(event.type)
print(stream.text)  # full accumulated text

Reply to a run

for event in client.runs.reply(run.id, message="also break it down by region"):
    print(event.type, event.raw)

# or block until complete
result = client.runs.reply_and_wait(run.id, message="also break it down by region")

Stream text only

for chunk in client.runs.stream_text(message="summarize inbox"):
    print(chunk, end="")

Need the run ID or accumulated text after? Use iter_text() instead:

with client.runs.create(message="summarize inbox") as stream:
    for chunk in stream.iter_text():
        print(chunk, end="", flush=True)
print(stream.run_id, stream.text)

Detect a failed stream

A run can fail mid-stream (expired credential, model rate limit, quota). The default iter_text() / stream.text path drops error events, so either opt into raising or check after iterating:

# Raise RunFailedError if the run fails mid-stream
for event in client.runs.create(message="...", raise_on_error=True):
    ...

# Or check without raising
with client.runs.create(message="...") as stream:
    for chunk in stream.iter_text():
        print(chunk, end="")
    if stream.has_errors:
        print("run failed:", stream.errors)

Resume a dropped stream

If the connection drops mid-run (proxy idle-timeout, network blip), rejoin with the run_id captured from the metadata event. runs.stream(run_id) replays the run's full history then live deltas, so reset any local accumulation on reconnect:

stream = client.runs.create(message="long autonomous task")
run_id = None
try:
    for event in stream:
        run_id = stream.run_id
        ...
except Exception:  # connection dropped mid-run
    if run_id:
        for event in client.runs.stream(run_id):  # re-attach and replay
            ...

The server emits a 15s keepalive on the streaming path so a long-silent tool call doesn't trip the read timeout; raise it for very long runs with M8tes(timeout=...).

Human-in-the-loop

Pass callbacks to wait(). Approval pauses are handled inline: Use PermissionMode constants to avoid string typos.

from m8tes import PermissionMode

run = client.runs.create(
    message="draft and send the weekly report",
    human_in_the_loop=True,
    permission_mode=PermissionMode.APPROVAL,
    task_setup_tools=False,  # keep this run limited to public tools only
    stream=False,
)
run = client.runs.wait(
    run.id,
    on_approval=lambda req: "allow",
    on_question=lambda req: {"Which channel?": "#general"},
)
print(run.output)

Or create and wait in a single call:

run = client.runs.create_and_wait(
    message="draft and send the weekly report",
    human_in_the_loop=True,
    permission_mode=PermissionMode.APPROVAL,
    on_approval=lambda req: "allow",
)

Low-level control

pending = client.runs.permissions(run.id)
client.runs.approve(run.id, request_id="req_123", decision="allow")
client.runs.answer(run.id, answers={"Which channel?": "#general"})

Switch permission mode on an existing run

run = client.runs.update_permission_mode(run.id, permission_mode=PermissionMode.APPROVAL)
print(run.permission_mode)  # "approval"

Switch mode while the run is still active, including awaiting_approval. Switching to PermissionMode.AUTONOMOUS auto-approves pending tool approval requests and resumes a paused tool approval run. AskUserQuestion and plan approvals still wait for client.runs.answer().

Computer use

When your account has sandbox execution enabled, agents run inside a full Linux desktop. No changes to your code — you get the same run API. The agent gains three extra tools automatically: computer (mouse/keyboard/screenshots), bash (shell), and str_replace_based_edit_tool (file editing).

with client.runs.create(
    agent_id=...,
    message="open chromium, go to example.com, and return the page title",
) as stream:
    for event in stream:
        if event.type == "tool_result":
            for block in event.content or []:
                if block.get("type") == "image":
                    # base64 PNG screenshot after each desktop action
                    screenshot_data = block["source"]["data"]
        if event.type == "text-delta":
            print(event.delta, end="")

Extra events in the stream:

Triggers

# schedule — every weekday at 9am (shortcut on tasks.create, no separate call needed)
task = client.tasks.create(agent_id=..., instructions="...", schedule="0 9 * * 1-5")

# webhook — POST to a URL to trigger runs
task = client.tasks.create(agent_id=..., instructions="...", webhook=True)
print(task.webhook_url)  # POST here to trigger (shown once)

# email — give the agent an inbox at creation time
mate = client.agents.create(name="inbox bot", email_inbox=True)
print(mate.email_address)  # forward emails here

# iMessage — route one BlueBubbles chat to an agent
messages_bot = client.agents.create(
    name="messages bot",
    inbound_imessage_enabled=True,
    imessage_chat_guid="iMessage;-;+15551231234",
)
print(messages_bot.imessage_chat_guid)  # use a dedicated 1:1 chat unless group access is intended

# on demand — run a saved task directly
for event in client.tasks.run(task.id):
    print(event.type, event.raw)

Multi-tenancy

Give each user their own AI agent with isolated memory, tools, and permissions.

# create a user profile
client.users.create(user_id="cust_123", name="Acme Corp", email="admin@acme.com")

# give them their own agent
bot = client.agents.create(
    name="acme assistant",
    tools=["gmail", "slack"],
    user_id="cust_123",
)

# seed their memory
client.memories.create(user_id="cust_123", content="prefers email over slack")

# pre-approve tools
client.permissions.create(user_id="cust_123", tool="gmail")

# run on their behalf — memory, permissions, history, and internal management tools all scoped
run = client.runs.create_and_wait(
    agent_id=bot.id,
    message="check inbox for urgent items",
    user_id="cust_123",
)

The same rule applies to saved tasks and follow-up runs:

task = client.tasks.create(
    agent_id=bot.id,
    instructions="review urgent inbox items",
)

# inherits cust_123 from the scoped agent
run = client.tasks.run(task.id, stream=False)
assert run.user_id == "cust_123"

Apps & connections

Inspect the app catalog first, then use the helper that matches the app's auth type.

apps = client.apps.list(user_id="cust_123")
for app in apps.data:
    print(app.name, app.auth_type, app.connected)

# OAuth app
start = client.apps.connect_oauth(
    "gmail",
    redirect_uri="https://app.example.com/oauth/callback",
    user_id="cust_123",
)
print(start.authorization_url)

# after your redirect handler gets the callback
client.apps.connect_complete("gmail", start.connection_id, user_id="cust_123")

# API key app
client.apps.connect_api_key("gemini", api_key="sk_live_...", user_id="cust_123")
client.apps.disconnect("gemini", user_id="cust_123")

# Platform-provisioned app (auth_type "platform_provisioned", e.g. twilio):
# the platform allocates a dedicated resource (a phone number) for you.
result = client.apps.provision("twilio", user_id="cust_123")
print(result.phone_number)  # "+15551234567"
client.apps.release("twilio", user_id="cust_123")  # release it back

Recursive Teams

Organize Mates into a hierarchy and grant a role across one subtree:

marketing = client.groups.create(name="Marketing")
paid_ads = client.groups.create(name="Paid Ads", parent_id=marketing.id)
google = client.groups.create(name="Google", parent_id=paid_ads.id)
client.agents.update(agent.id, group_id=google.id)

invite = client.groups.invite(marketing.id, email="ada@example.com", role="runner")
members = client.groups.members(google.id)  # includes inherited roles
client.groups.update_member(marketing.id, members.data[0].member_id, role="editor")

New invitations default to editor; choose viewer for read access or runner for read plus run/chat/reply/cancel/approve. Editors can also edit Mates, tasks, documents, and work inside the subtree. Existing grants remain viewer. Roles inherit through child Teams. Editors cannot manage members or roles, self-escalate, reorganize Teams, or move a Mate out of the shared scope.

Team sharing does not make someone an organization member. Runs use the shared Mate owner's bound tools and billing while auditing the human requester separately; credentials stay opaque. Group CRUD accepts user_id for end-user isolation, while membership and invitation methods do not. groups.share() remains the separate legacy bulk operation for direct Mates' visibility.

Resources

Pagination

# standard page
page = client.runs.list(limit=50)
for run in page.data:
    print(run.id, run.status)

# auto-paginate through all results
for run in client.runs.list(user_id="customer_123").auto_paging_iter():
    print(run.id, run.status)

Webhooks

# register an endpoint
hook = client.webhooks.create(
    url="https://example.com/hook",
    events=["run.completed", "run.failed"],
)
secret = hook.secret  # save this — only shown once

# verify incoming webhooks (e.g. in Flask/FastAPI)
from m8tes import Webhooks

is_valid = Webhooks.verify_signature(
    body=request.body,
    headers=dict(request.headers),
    secret=secret,
)

Files

files = client.runs.list_files(run_id=42)
for f in files:
    print(f.name, f.size)

content = client.runs.download_file(run_id=42, filename="report.csv")

Error handling

from m8tes import M8tes, NotFoundError, RateLimitError, AuthenticationError

try:
    client.agents.get(999)
except NotFoundError:
    print("agent not found")
except RateLimitError as e:
    print(f"rate limited, retry after {e.retry_after}s")
except AuthenticationError:
    print("invalid API key")

Run-level failures

Exceptions above cover problems reaching the API. A run can also fail upstream — an expired Claude credential, an exhausted plan quota, a model rate limit. The HTTP call succeeds, so no exception is raised, but the run carries the failure: status is "completed", the message is in run.output, and run.error_code holds a machine-readable class (e.g. oauth_revoked, subscription_quota_exhausted, rate_limited). Check error_code before trusting output:

run = client.runs.create_and_wait(agent_id=mate.id, message="...")
if run.error_code:
    print(f"run failed upstream: {run.error_code} — {run.output}")
else:
    print(run.output)

Configuration

client = M8tes(api_key="m8_...", timeout=300)  # custom timeout in seconds

CLI

m8tes auth login                    # authenticate
m8tes auth usage                    # account limits and current usage
m8tes apps connect-api-key gemini KEY
m8tes agent create --non-interactive --name "messages bot" --tools gmail --instructions "Help via iMessage" --enable-imessage --imessage-chat-guid "iMessage;-;+15551231234"
m8tes run set-permission-mode 42 approval
m8tes agent task ID "message"       # run a task
m8tes agent chat ID                 # interactive chat

mate is a permanent alias (m8tes mate task … still works).

m8tes run set-permission-mode also works while a run is paused. Switching to autonomous resumes pending tool approvals, but AskUserQuestion still waits for an explicit answer.

See CLI documentation for all commands and options.

Links

• Documentation
• Developer hub
• Examples
• Changelog
• PyPI
• support@m8tes.ai

Contributing

Bug reports and feature requests are welcome — open an issue; we review weekly. We don't currently accept external pull requests: this repo is synced from our internal monorepo, so changes land through our own pipeline. If something blocks you, an issue (or support@m8tes.ai) is the fastest path to a fix.

License

MIT — see LICENSE for details. The m8tes name and logo are trademarks of m8tes; the MIT license does not grant trademark rights.