Python SDK

Execute Fruxon agents programmatically from Python — one-shot, streaming, multi-turn — plus full authoring, agent-network, and observability APIs.

The Fruxon Python SDK is the programmatic surface for the whole platform: execute deployed agents in your application, and author everything the CLI does — drafts, revisions, integrations, tools, triggers, assets, and the agent network. For terminal usage, see the CLI guide.

The CLI and SDK ship in the same package (pip install fruxon). The SDK is the only supported way to run a deployed agent in production — the CLI deliberately runs only drafts (Origin=TEST).

Install

pip install fruxon

Requires Python 3.10+.

Authentication

You need a Fruxon bearer token and your organization identifier. Generate a personal token or service-account token from Settings -> API Access.

Pass the token and org to the constructor (both are keyword-only):

from fruxon import FruxonClient

client = FruxonClient(
    token="...",
    org="acme-corp",
)

Unlike the CLI, the SDK does not read FRUXON_TOKEN / FRUXON_ORG from the environment for you. If you want that behavior, read them yourself with os.environ.

Client options

Parameter	Type	Default	Description
`token`	`str`	required	Fruxon token (`fx_pat_...` or `fx_sat_...`)
`org`	`str`	required	Organization identifier
`base_url`	`str`	`"https://api.fruxon.com"`	Override for self-hosted or staging environments
`timeout`	`float`	`120.0`	Per-request timeout in seconds
`actor`	`str \| None`	autodetected	Attribution string sent as `X-Fruxon-Actor` for audit trails. Auto-detected from the environment when omitted.

Execute an agent

result = client.execute(
    "support-agent",
    parameters={"question": "How do I reset my password?"},
)

print(result.response)
print(f"{result.trace.duration}ms · ${result.trace.total_cost:.4f}")
print(result.execution_record_id)

`execute()` parameters

Parameter	Type	Description
`agent`	`str`	Agent ID (positional).
`parameters`	`dict`	Inputs matching the agent's entry-point schema.
`session_id`	`str`	Continue a multi-turn conversation (see below).
`attachments`	`list`	File attachments to include with the request.
`chat_user`	`dict`	Identify the end user (for traceability and access control).
`environment_slug`	`str`	Attribute the run to an end-customer environment for per-environment cost and analytics. See Environments.

Result shape

execute() returns an ExecutionResult:

Field	Type	Description
`response`	`str`	The agent's text response.
`session_id`	`str \| None`	Pass back to continue this conversation.
`execution_record_id`	`str`	Unique ID — use with `fruxon agents executions trace` or `get_execution_record()`.
`trace`	`ExecutionTrace`	Cost, duration, token counts.
`links`	`list`	Related links returned by the agent.

The trace object carries agent_id, agent_revision, duration (ms), the cost breakdown (input_cost, output_cost, total_cost, cached_cost, cache_write_cost, web_search_cost, thinking_cost, provider_reported_cost), and token counts (input_tokens, output_tokens, cached_tokens, cache_write_tokens, web_search_calls).

Multi-turn conversations

Thread the returned session_id into the next call:

first = client.execute(
    "support-agent",
    parameters={"question": "What's your return policy?"},
)

follow_up = client.execute(
    "support-agent",
    parameters={"question": "What about international orders?"},
    session_id=first.session_id,
)

Streaming

For incremental text as the agent generates it:

for chunk in client.stream_text(
    "support-agent",
    parameters={"question": "Tell me about your service"},
):
    print(chunk, end="", flush=True)

For the full typed event stream — text chunks, tool calls, tool results, step traces, terminal events:

for event in client.stream("support-agent", parameters={"question": "Hi"}):
    if event.event == "text":
        print(event.data["chunk"], end="", flush=True)
    elif event.event == "tool_call":
        print(f"\n[tool call] {event.data}")
    elif event.event == "tool_result":
        print(f"\n[tool result] {event.data}")
    elif event.event == "done":
        trace = event.data.get("trace", {})
        print(f"\n[done] {trace.get('duration')}ms")

Each StreamEvent has an event string and a data dict. Event types: text, tool_call, tool_result, usage, status, step_trace, error, done. stream() and stream_text() accept the same parameters as execute().

Test a draft flow

Run the agent's saved server-side draft without publishing it — the programmatic side of fruxon agents draft run. Save the draft first (in Studio or via put_draft), then run it against a base_revision. The request is an execution body (parameters, attachments, sessionId, mode); masked secrets resolve from the base revision's configs.

request = {"parameters": {"query": "hello"}}

# Sync — base_revision is a required positional. Same ExecutionResult shape as execute()
result = client.test("support-agent", 3, request)
print(result.response)

# Streaming — same StreamEvent shape as stream()
for event in client.stream_test("support-agent", 3, request):
    if event.event == "text":
        print(event.data["chunk"], end="", flush=True)

The SDK posts request verbatim — it deliberately doesn't model the broad, fast-evolving AgentTestRequest schema. Requires Editor access to the agent (the token's owning user or service account must have sufficient access).

Discovery

List every agent in your org, inspect one by ID, or fetch the input parameters a revision expects:

for agent in client.list_agents():
    print(agent.id, agent.display_name, agent.current_revision)

one = client.get_agent("support-agent")

params = client.get_agent_parameters(
    "support-agent",
    revision=one.current_revision,
)

list_agents accepts search=, tags=, and all_pages=. The narrow Agent dataclass exposes id, display_name, description, enabled, current_revision, tags.

Authoring: drafts & revisions

The full author → mint → deploy loop. A draft is the mutable working copy; a revision is an immutable snapshot you deploy.

# Drafts
draft = client.get_draft("support-agent", base_revision=7)
client.put_draft("support-agent", base_revision=7, head={...})   # save edits
client.undo_draft("support-agent", base_revision=7)
client.redo_draft("support-agent", base_revision=7)
client.reset_draft("support-agent", base_revision=7)
client.delete_draft("support-agent", base_revision=7)
client.list_drafts("support-agent")

# Score a draft against a golden dataset (expensive — runs the flow per sample)
run = client.evaluate_draft("support-agent", base_revision=7, dataset_id="...")

# Mint & deploy
client.create_agent({"id": "support-agent", "displayName": "Support"})
rev = client.create_revision("support-agent", {...})    # body = CreateAgentRevision
client.deploy_revision("support-agent", rev_number)
client.get_revision("support-agent", 42)                # fork from a known-good revision
client.delete_agent("support-agent")

The write methods (put_draft, create_revision, …) post the request body verbatim and accept an optional if_match for optimistic concurrency. subscribe_draft(agent, base_revision) opens a live stream of edits from other sessions.

Observability

# List past executions — filter by status / origin / revision / time window
for rec in client.list_execution_records("support-agent", status="FAILED", page_size=20):
    print(rec.id, rec.status, rec.total_cost)

record = client.get_execution_record("support-agent", "rec-abc123")
print(record.status, record.total_cost)

# Full step-by-step trace, and the agent's final output, as raw dicts
trace = client.get_execution_trace("support-agent", "rec-abc123")
output = client.get_execution_result("support-agent", "rec-abc123")

The ExecutionRecord dataclass carries id, agent_id, agent_revision, status, start_time, end_time, total_cost, and the token/cost breakdown. The record ID is printed at the end of every execute() and surfaced by fruxon agents draft run in the terminal.

Cost & budgets

client.get_test_cost("support-agent", revision=12)        # cumulative dev spend
client.list_agent_budgets("support-agent")
client.get_agent_budget("support-agent", origin="TEST")
client.upsert_agent_budget("support-agent", amount=50, origin="TEST", enforce_limit=True)
client.delete_agent_budget("support-agent", origin="TEST")

Human approvals

Operate a step's human-in-the-loop gate — the decisions an agent is blocked on:

for approval in client.list_approvals("support-agent", status="pending"):
    print(approval.id, approval.gate_type, approval.step_identifier)

client.get_approval("support-agent", "appr-1")
client.respond_approval("support-agent", "appr-1", text="approved")
client.cancel_approval("support-agent", "appr-1", reason="stale")

ApprovalRequest carries id, status, gate_type, step_identifier, tool_key, proposed_parameters, authorized_approvers, decision, and timestamps.

The agent network

Fruxon agents consult one another and route to people. The SDK exposes the whole fabric.

Participants

for p in client.list_participants():
    print(p.id, p.kind, p.display_name)

client.get_participant("p_123")
client.create_participant({...})
client.update_participant("p_123", {...})
client.enable_participant("p_123")
client.disable_participant("p_123")
client.delete_participant("p_123")

# Bind a participant to an agent's consult roster, with a policy
client.bind_participant("p_123", "support-agent")
client.set_consult_roster("p_123", "support-agent", {...})   # roles, urgency, response policy
client.unbind_participant("p_123", "support-agent")
client.list_consult_roster(agent="support-agent")

Capabilities & consult pins

for cap in client.list_capabilities():
    print(cap.id, cap.name, cap.area)

client.create_capability({...})
client.update_capability("cap-1", {...})
client.delete_capability("cap-1")

# Deterministic capability → participant routing overrides
client.list_consult_pins(agent="support-agent")
client.create_consult_pin({...})
client.delete_consult_pin("pin-1")

Topics, inbox & memory

# Conversation topics
for t in client.list_topics("support-agent", state="open"):
    print(t.id, t.goal, t.state)
client.search_topics("support-agent", "refund", limit=10)
client.get_topic("support-agent", "topic-1")
client.list_topic_messages("support-agent", "topic-1")

# The agent's attention landscape (focal / home / suspended topics)
client.get_inbox("support-agent", participant="p_123")

# Long-term memory
for m in client.list_memories("support-agent", subject="p_123"):
    print(m.subject, m.title)
client.list_memory_subjects("support-agent")
client.get_memory("support-agent", "mem-1")
client.forget_subject("support-agent", "p_123")

Channels & endpoints

client.list_channel_bindings("support-agent")        # how it receives/sends inbound
client.list_messaging_endpoints("support-agent")     # resolved endpoints + bot identity

Sandbox

Drive a full agent-network simulation (turns, trigger fires, consults, approvals) without touching production:

session = client.open_sandbox_session("support-agent")
client.send_sandbox_turn(session, participant="p_123", text="hi")
client.fire_sandbox_trigger(session, trigger="trg-1", sample_payload={...})
client.resolve_sandbox_trigger_input(session, trigger="trg-1", sample_payload={...})
client.answer_sandbox_consult(session, operation_id="op-1", text="approved")
for event in client.stream_sandbox(session):
    ...
client.list_sandbox_messages(session)
client.close_sandbox_session(session)

Integrations & tools

The building blocks an agent draws on: an integration is a connection to an external service, and tools are the callable capabilities defined under it.

# Integrations
for integ in client.list_integrations(search="git", types=["CUSTOM"], tags=["ci"]):
    print(integ.id, integ.type, integ.tags)

client.get_integration("github")
client.create_integration({"id": "github", "displayName": "GitHub", "configMetadata": {...}})
client.update_integration("github", {"displayName": "GitHub (renamed)"})

# A failed verification is a result, not an exception
check = client.verify_integration("github", {"authMetadataId": "...", "authConfig": {...}})

# OAuth: mint a connect link for a human to click; discover auth methods / event types
link = client.authorize_integration("github", {...})
client.list_integration_auth_methods("github")
client.list_integration_triggers("github")      # event types this integration can fire on
client.list_integration_configs("github")

# MCP server exposure
client.get_mcp_config("github")
client.update_mcp_config("github", {...})
client.rotate_mcp_key("github")

# Tools — always integration-scoped: (integration, tool) is the composite key
for tool in client.list_tools("github", types=["API"]):
    print(tool.id, tool.tool_type, tool.action_type)

client.get_tool("github", "list_commits")
client.create_tool("github", {"id": "list_commits", "integrationId": "github", "descriptor": {...}})
client.update_tool("github", "list_commits", {...})
client.update_tool("github", "run_report", {...}, python=True)   # Python-script tools
client.delete_tool("github", "list_commits")

# Test-run a tool before wiring it into an agent — returns the raw {response}
client.test_tool("github", {"descriptor": {...}, "parameters": {"repo": "fruxon-sdk"}})

list_* / get_* return the narrow Integration / Tool dataclasses. The write methods post the request body verbatim — the structured configMetadata / descriptor payloads aren't modelled by the SDK — and return the same narrow view of the result. verify_integration, test_tool, and the MCP/auth/config helpers return raw dicts.

Triggers

A trigger is a scheduled or event source that fires an agent. Discover, author, and bind them:

for t in client.list_triggers():
    print(t.id, t.kind, t.event_type, t.agent_ids)

client.get_trigger("trg-1")                 # → TriggerDetail (schedule, bindings, filters)
client.create_trigger({...})
client.update_trigger("trg-1", {...})
client.delete_trigger("trg-1")
client.fire_trigger("trg-1", sample_payload={...})   # simulate a fire

# A bound trigger is what gives an agent an inbound path
client.bind_trigger("trg-1", "support-agent")
client.unbind_trigger("trg-1", "support-agent")

Knowledge assets (RAG)

# Upload a file and vectorize it for retrieval, then wait for ingestion
asset = client.create_asset_from_file("handbook.pdf", vectorize=True)
client.wait_for_asset(asset.id)

for a in client.list_assets():
    print(a.id, a.asset_type, a.status, a.vectorized)

client.get_asset("asset-1")
client.update_asset("asset-1", {...})
client.delete_asset("asset-1")
client.list_asset_operations("asset-1")
client.list_asset_embedding_models()
client.list_asset_supported_types()

Object storage

Upload local files (e.g. images for OCR / vision steps) and get back a link to drop into an agent input:

link = client.upload_file("invoice.png")          # → dict: fileId, contentType, promptLink, …
data, content_type = client.download_file(link["fileId"])   # → (bytes, content_type)

Secrets & metrics

# Secrets — metadata only, never values
for s in client.list_secrets():
    print(s.id, s.key, s.sensitivity)
client.get_secret("secret-1")
client.list_secret_grants("secret-1")

# Evaluation metrics — ids LLM-judge steps and eval runs reference
client.list_eval_metrics()
client.judge_output("support-agent", output="...", metric_ids=["helpfulness"], input="...")

Datasets & evaluation runs

client.list_datasets("support-agent")
client.get_dataset("support-agent", "dataset-1")
client.get_evaluation_run("support-agent", "run-1")

LLM providers & skills

client.list_llm_providers()
client.get_llm_provider("anthropic")
client.list_llm_provider_configs("anthropic")
client.get_llm_provider_config("anthropic", "config-1")

client.list_skills()
client.get_skill("fruxon-create-integration")

Token management

client.list_tokens()
client.generate_token({...})                     # mint a scoped token
client.rotate_token("tok-1", overrides={...})
client.revoke_token("tok-1")
client.list_token_audit_events("tok-1", page_size=50)
client.get_current_token()                       # introspect the token in use
client.list_token_scopes()                       # every scope the server mints against

Error handling

All API errors derive from FruxonError. Branch on the subclasses to give users actionable messages:

from fruxon import FruxonClient
from fruxon.exceptions import (
    AuthenticationError,
    ForbiddenError,
    NotFoundError,
    ValidationError,
    FruxonConnectionError,
    FruxonAPIError,
    FruxonError,
)

client = FruxonClient(token="...", org="acme-corp")

try:
    result = client.execute("my-agent", parameters={"input": "Hello"})
except AuthenticationError:
    print("Token was rejected — generate or rotate one in Settings.")
except NotFoundError:
    print("Agent not found in this organization.")
except ValidationError as e:
    print(f"Bad request: {e}")
except FruxonConnectionError:
    print("Could not reach the Fruxon API.")
except FruxonError as e:
    print(f"Unexpected error: {e}")

Exception	When
`AuthenticationError`	Invalid or missing Fruxon token (401).
`ForbiddenError`	Insufficient permissions (403).
`NotFoundError`	Agent or organization not found (404).
`ValidationError`	Invalid parameters / missing required fields (400 / 422).
`FruxonConnectionError`	Network error reaching the API.
`FruxonAPIError`	Base class for HTTP-level errors (catch this for any API failure).
`FruxonError`	Top-level base class for all SDK exceptions.

Self-hosted / staging

Point at an alternate API host:

client = FruxonClient(
    token="...",
    org="acme-corp",
    base_url="https://api.staging.fruxon.com",
)