Skip to main content

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​

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
)
Env variableDescriptionDefault
FLYMYAI_API_KEYAPI key (used when api_key is omitted)—
FLYMYAI_DSNBase URL overridehttps://backend.flymy.ai

The client exposes four resource namespaces:

NamespaceWhat it does
client.agentsCreate, list, update, delete, and run agents
client.runsStart runs, poll for results, stream events, follow up
client.toolsBrowse the tool catalog, add and configure MCP tools
client.compilationsCompile 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)
)

print(agent.id)     # UUID string
print(agent.goal)   # alias for user_prompt
print(agent.status) # "draft"
ParameterTypeRequiredDescription
namestrYesHuman-readable name
goalstrYesInstructions (supports Handlebars templating)
toolslist[int]NoTool IDs to attach
statusstrNodraft (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:

TypeMeaning
declared_functionsAgent declared the tools it will use
tool_calledA tool was invoked
tool_call_exceptionA tool call failed
task_cancelledThe 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​

Turn a successful agent run into a standalone Python script.

# Compile
compilation = client.compilations.compile(execution_id=run.id)

# Wait for compilation (poll)
import time
while compilation.status in ("pending", "compiling"):
    time.sleep(2)
    compilation = client.compilations.get(compilation.id)

print(compilation.script_code)  # generated Python

# Execute
result = client.compilations.run(compilation.id)

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."}
ExceptionWhen
FlyMyAIAgentErrorAny non-2xx API response (has .status_code and .response_body)
TimeoutErrorruns.wait() or runs.stream_events() exceeded timeout
ValueErrorMissing api_key at initialization

Type reference​

Agent​

FieldTypeDescription
uuid / idstrAgent UUID
namestrAgent name
user_prompt / goalstrInstructions
available_toolslistTool IDs (list view) or tool objects (detail view)
statusAgentStatusdraft, initialization_required, active, archived
all_tools_configuredboolWhether all attached tools are configured
generated_pipelinedictAuto-generated execution pipeline

Run / RunDetail​

FieldTypeDescription
idintExecution ID
statusExecutionStatuspending, running, completed, failed, cancelled
agent_result / outputdictFinal output
errorstrError message (if failed)
messageslist[dict]Full conversation history
logslist[ExecutionLog]Step-by-step logs (RunDetail only)
is_terminalboolWhether the run has finished
original_promptstrThe prompt used for this run

ExecutionLog​

FieldTypeDescription
idintLog entry ID
typestrdeclared_functions, tool_called, tool_call_exception, task_cancelled
messagestrHuman-readable description
datadictStructured payload (arguments, results, etc.)

Tool​

FieldTypeDescription
idintTool ID
mcp_tool / namestrTool identifier (e.g. "github", "tavily")
is_configuredboolAll configuration steps complete
is_activeboolTool enabled
next_configuration_stepdictNext step to complete (or None)
user_configdictCurrent configuration values

AvailableTool​

FieldTypeDescription
namestrTool identifier
typestroauth, custom, composio, custom_class
titlestrHuman-readable name
descriptionstrShort description
categorieslist[str]Categories (e.g. ["Development"])

Compilation​

FieldTypeDescription
idintCompilation ID
statusstrpending, compiling, compiled, running, completed, failed
script_codestrGenerated Python script
resultdictExecution result
errorstrError message (if failed)