Python SDK

Everything you need to build, run, and refine agents from Python.

Installation

pip install flymyai

Requires Python 3.8+. The agents module ships inside the flymyai package.

30-second example

from flymyai import AgentClient

client = AgentClient(api_key="fly-***")

# 1. Add a tool
tool = client.tools.create(mcp_tool="tavily")

# 2. Create an agent
agent = client.agents.create(
    name="Web Researcher",
    goal="Search the web for {{ topic }} and return a concise summary with sources.",
    tools=[tool.id],
)

# 3. Run & wait
run = client.runs.create(agent_id=agent.id)
result = client.runs.wait(run.id)
print(result.output)

---

Client

Sync

from flymyai import AgentClient

client = AgentClient(
    api_key="fly-***",       # or set FLYMYAI_API_KEY env var
    base_url="...",           # default: https://backend.flymy.ai
    timeout=60.0,             # request timeout in seconds
)

Async

from flymyai import AsyncAgentClient

async with AsyncAgentClient(api_key="fly-***") as client:
    agent = await client.agents.create(name="Bot", goal="Help users")
    run = await client.runs.create(agent_id=agent.id)
    result = await client.runs.wait(run.id)

Env variable	Description	Default
FLYMYAI_API_KEY	API key (used when api_key is omitted)	-
FLYMYAI_AGENTS_BASE_URL	Base URL override for the agents API	https://backend.flymy.ai

Two distinct base URLs

The agents API (AgentClient) and the model-inference API (FlyMyAI, flymyai.run) live on different hosts and have separate env vars:

Client	Default base URL	Env var override
AgentClient (this page)	https://backend.flymy.ai	FLYMYAI_AGENTS_BASE_URL
FlyMyAI (model inference)	https://api.flymy.ai/	FLYMYAI_DSN

Setting one does not affect the other, so you can point each client at staging independently.

The client exposes four resource namespaces:

Namespace	What it does
client.agents	Create, list, update, delete, and run agents
client.runs	Start runs, poll for results, stream events, follow up
client.tools	Browse the tool catalog, add and configure MCP tools
client.compilations	Compile agent runs into standalone scripts

---

client.agents

Create

agent = client.agents.create(
    name="Lead Profiler",
    goal="""Research {{ name }} at {{ company }}.
    Return background, recent initiatives, and recommended outreach angle.""",
    tools=[web_search_id, browse_id],  # tool IDs (integers)
    input_schema={
        "type": "object",
        "properties": {
            "name": {"type": "string"},
            "company": {"type": "string"},
        },
        "required": ["name", "company"],
    },
    input_description="Name and company of the lead to research.",
    output_schema={
        "type": "object",
        "properties": {
            "background": {"type": "string"},
            "outreach_angle": {"type": "string"},
        },
        "required": ["background", "outreach_angle"],
    },
    output_description="A short background blurb and a recommended outreach angle.",
)

print(agent.id)     # UUID string
print(agent.goal)   # alias for user_prompt
print(agent.status) # "draft"

Parameter	Type	Required	Description
name	str	Yes	Human-readable name
goal	str	Yes	Instructions (supports Handlebars templating)
tools	list[int]	No	Tool IDs to attach
mcp_servers	list[int]	No	Custom MCP server IDs to attach
input_schema	dict	No	JSON Schema for runtime variables (required if goal has placeholders)
input_description	str	No	Plain-text description of what the user provides; authoritative scope at freeze time
output_schema	dict	No	JSON Schema the agent must produce as its final result
output_description	str	No	Plain-text description of the result; pairs with output_schema to bound the canonical pipeline
status	str	No	draft (default) or active

List / Get / Update / Delete

# List all agents
agents = client.agents.list()

# Get with full details (nested tool objects)
agent = client.agents.get("a1b2c3d4-...")

# Partial update - only send the fields you want to change
agent = client.agents.update("a1b2c3d4-...",
    goal="Updated instructions.",
)

# Soft-delete (archive)
client.agents.delete("a1b2c3d4-...")

Run

Start the agent loop. Returns immediately - the agent executes asynchronously on the server.

run = client.agents.run("a1b2c3d4-...")
print(run.id)      # execution ID (int)
print(run.status)  # "pending" or "running"

tip

client.runs.create(agent_id=...) is an alias for client.agents.run(...). Use whichever reads better in your code.

---

client.runs

Create a run

run = client.runs.create(agent_id=agent.id)

Get run details

run = client.runs.get(42)
print(run.status)   # "completed"
print(run.output)   # the agent's result (dict)
print(run.error)    # None or error message

for log in run.logs:
    print(f"[{log.type}] {log.message}")

Wait for completion

Polls the server until the run finishes. Returns the final RunDetail.

result = client.runs.wait(run.id, timeout=300, poll_interval=2.0)

if result.status == "completed":
    print(result.output)
else:
    print(f"Run ended with status: {result.status}")
    print(f"Error: {result.error}")

Raises TimeoutError if the run doesn't finish in time.

Stream events

Watch execution step-by-step. Yields ExecutionLog objects as they appear.

run = client.runs.create(agent_id=agent.id)

for event in client.runs.stream_events(run.id, timeout=600):
    if event.type == "tool_called":
        print(f"  Tool: {event.message}")
    elif event.type == "tool_call_exception":
        print(f"  Error: {event.message}")
    else:
        print(f"  [{event.type}] {event.message}")

Event types:

Type	Meaning
declared_functions	Agent declared the tools it will use
tool_called	A tool was invoked
tool_call_exception	A tool call failed
task_cancelled	The run was cancelled

Cancel

client.runs.cancel(42)

Follow up

Append a user message and restart the agent loop on the same execution:

run = client.runs.append_message(run.id, text="Also check their LinkedIn profile.")
result = client.runs.wait(run.id)

---

client.tools

Tools are MCP integrations available on the FlyMy.AI platform. You configure them once, then attach to any agent.

Browse the catalog

catalog = client.tools.available()
for t in catalog:
    print(f"{t.name:20s} {t.type:15s} {t.title}")
    # github               composio        Github
    # tavily               custom_class    Tavily
    # telegram-mcp         oauth           Telegram Mcp

Add and configure a tool

# Add
tool = client.tools.create(mcp_tool="github")
print(tool.is_configured)  # False

# Walk through configuration steps
while not tool.is_configured:
    step = tool.next_configuration_step
    if not step:
        break
    print(f"Step: {step['description']}")
    # → "Provide your GitHub token"

    tool = client.tools.provide_config(
        tool.id,
        user_response={"GITHUB_TOKEN": "ghp_..."},
    )

print(tool.is_configured)  # True

Call a tool directly

For custom_class tools, you can invoke actions without running a full agent:

result = client.tools.call(
    tool.id,
    action="search",
    arguments={"query": "FlyMyAI agents"},
)
print(result)

Other operations

tools = client.tools.list()                            # your configured tools
tool = client.tools.get(7)                              # by ID
tool = client.tools.update(7, user_config={"key": "new_value"})  # merge config
client.tools.delete(7)                                  # remove

---

client.compilations

Once a chat session has done the right thing, you can freeze it: the backend reads the chat, the agent's input_description / output_description, and the schemas, then distills the canonical input → output pipeline into a Markdown instruction. The frozen instruction can be re-run later with fresh variables — fast, deterministic, no re-exploration.

Freeze and run end-to-end

# 1. The agent is happy with the last run.
run = client.runs.get(run_id)

# 2. Freeze it - returns immediately, then polls until COMPILED.
compilation = client.agents.compile_from_run(run.id, timeout=120)
print(compilation.instruction_md)   # the frozen Markdown plan

# 3a. Run without variables (agent has no input_schema):
result_run = client.compilations.run_instruction_and_wait(compilation.id)
print(result_run.output)

# 3b. Run with variables (agent has an input_schema):
from flymyai import VariablesValidationError

try:
    result_run = client.compilations.run_instruction_and_wait(
        compilation.id,
        variables={"website_url": "https://example.com"},
    )
except VariablesValidationError as err:
    print(err.field_errors)
    raise

print("Status:", result_run.status)
print("Output:", result_run.output)            # shaped by output_schema
print("Score:", result_run.output["score"])    # individual fields

Lower-level building blocks

Method	What it does
client.agents.freeze(execution_id)	POST /api/v1/agents/compilations/freeze-instruction/{id}/; returns immediately, status starts at pending
client.agents.compile_from_run(execution_id, timeout=120)	Freeze + poll until terminal (compiled / failed)
client.compilations.list()	Every compilation in the workspace
client.compilations.get(comp_id)	One compilation with instruction_md, status, error
client.compilations.run_instruction(comp_id, variables=...)	Dispatch a fresh execution from the frozen instruction; returns a new Run immediately
client.compilations.run_instruction_and_wait(comp_id, variables=..., timeout=600)	Same as above, plus polling until terminal
client.compilations.wait(comp_id, timeout=120)	Poll an existing compilation until terminal

Legacy: compile to Python

For workflows where you want a generated Python script instead of a Markdown plan, the older compile endpoint is still available:

compilation = client.compilations.compile(execution_id=run.id)
client.compilations.wait(compilation.id)        # poll until compiled
print(compilation.script_code)                  # generated Python
client.compilations.run(compilation.id)         # execute the script

Prefer freeze + run_instruction for new code — it carries the canonical scope from input_description / output_description and runs through the same agent loop with full tool support.

---

Training with patches

Improve your agent iteratively by evaluating output and patching its configuration.

Refine the goal

run = client.runs.create(agent_id=agent.id)
result = client.runs.wait(run.id)

# Output too verbose? Patch the instructions
client.agents.update(agent.id,
    goal="Research {{ topic }}. Be concise: max 3 paragraphs, bullet points preferred.",
)

Automated training loop

for i in range(5):
    run = client.runs.create(agent_id=agent.id)
    result = client.runs.wait(run.id)

    score = evaluate(result.output)  # your scoring function
    if score >= 0.8:
        break

See the Training with Patches guide for detailed patterns.

---

Async usage

Every method has an async counterpart:

import asyncio
from flymyai import AsyncAgentClient

async def main():
    async with AsyncAgentClient(api_key="fly-***") as client:
        agent = await client.agents.create(
            name="Researcher",
            goal="Find info about {{ topic }}.",
        )
        run = await client.runs.create(agent_id=agent.id)

        # Stream events
        async for event in client.runs.stream_events(run.id):
            print(f"[{event.type}] {event.message}")

        result = await client.runs.get(run.id)
        print(result.output)

asyncio.run(main())

---

Complete example

End-to-end: set up tools, create agent, run, stream, follow up, train.

from flymyai import AgentClient

client = AgentClient(api_key="fly-***")

# ── 1. Set up tools ─────────────────────────────────────────────
catalog = client.tools.available()
print("Available:", [t.name for t in catalog])

tool = client.tools.create(mcp_tool="tavily")
tool = client.tools.provide_config(
    tool.id,
    user_response={"TAVILY_API_KEY": "tvly-..."},
)

# ── 2. Create agent ─────────────────────────────────────────────
agent = client.agents.create(
    name="Research Assistant",
    goal="Research {{ topic }} thoroughly. Return a summary with key findings and sources.",
    tools=[tool.id],
)

# ── 3. Run & stream ─────────────────────────────────────────────
run = client.runs.create(agent_id=agent.id)
print(f"Run {run.id} started")

for event in client.runs.stream_events(run.id, timeout=600):
    print(f"  [{event.type}] {event.message}")

# ── 4. Get result ────────────────────────────────────────────────
result = client.runs.get(run.id)
if result.status == "completed":
    print("\nOutput:", result.output)

# ── 5. Follow up ────────────────────────────────────────────────
updated = client.runs.append_message(
    run.id, text="Now compare this with last year's trends."
)
final = client.runs.wait(updated.id, timeout=600)
print("\nUpdated:", final.output)

---

Error handling

from flymyai import AgentClient, FlyMyAIAgentError

client = AgentClient(api_key="fly-***")

try:
    agent = client.agents.get("nonexistent-uuid")
except FlyMyAIAgentError as e:
    print(e.status_code)     # 404
    print(e.response_body)   # {"detail": "Not found."}

Exception	When
FlyMyAIAgentError	Any non-2xx API response (has .status_code and .response_body)
TimeoutError	runs.wait() or runs.stream_events() exceeded timeout
ValueError	Missing api_key at initialization

---

Type reference

Agent

Field	Type	Description
uuid / id	str	Agent UUID
name	str	Agent name
user_prompt / goal	str	Instructions
input_schema	dict | None	JSON Schema for runtime variables
input_description	str	Plain-text description of the agent input
output_schema	dict | None	JSON Schema for the agent's final result
output_description	str	Plain-text description of the agent output
available_tools	list	Tool IDs (list view) or tool objects (detail view)
available_custom_mcp_servers	list[int]	Custom MCP server IDs attached
status	AgentStatus	draft, initialization_required, active, archived
all_tools_configured	bool	Whether all attached tools are configured
generated_pipeline	dict	Auto-generated execution pipeline
cron_schedule	str	Cron expression for scheduled runs (or "")
webhook_url	str | None	Webhook URL for run notifications

Run / RunDetail

Field	Type	Description
id	int	Execution ID
status	ExecutionStatus	pending, running, completed, failed, cancelled
agent_result / output	dict	Final output
error	str	Error message (if failed)
messages	list[dict]	Full conversation history
logs	list[ExecutionLog]	Step-by-step logs (RunDetail only)
is_terminal	bool	Whether the run has finished
original_prompt	str	The prompt used for this run

ExecutionLog

Field	Type	Description
id	int	Log entry ID
type	str	declared_functions, tool_called, tool_call_exception, task_cancelled
message	str	Human-readable description
data	dict	Structured payload (arguments, results, etc.)

Tool

Field	Type	Description
id	int	Tool ID
mcp_tool / name	str	Tool identifier (e.g. "github", "tavily")
is_configured	bool	All configuration steps complete
is_active	bool	Tool enabled
next_configuration_step	dict	Next step to complete (or None)
user_config	dict	Current configuration values

AvailableTool

Field	Type	Description
name	str	Tool identifier
type	str	oauth, custom, composio, custom_class
title	str	Human-readable name
description	str	Short description
categories	list[str]	Categories (e.g. ["Development"])

Compilation

Field	Type	Description
id	int	Compilation ID
status	str	pending, compiling, compiled, running, completed, failed
script_code	str	Generated Python script
result	dict	Execution result
error	str	Error message (if failed)