flymyai 1.2.0rc9

FlyMy.AI - the Agentic Cloud.

ᕦ[▀̿_▀̿]ᕤ ⚡ ◢▆[◉◡◉]▆◣ ⚡ o=[•_•]=o ⚡ ╾━[⊙▂⊙]━╼ ⚡ ¬[°□°]¬ ⚡ 凸[¬_¬]凸 ⚡ <|¤_¤|> ⚡ ʕ[•ᴥ•]ʔ ⚡ ╚[ʘᗜʘ]╝ ⚡ d[-_-]b ⚡ q[◔౪◔]p ⚡ \[T_T]/ ⚡ ✧[◕‿◕]✧ ⚡ =^.^=

FlyMy.AI is the Agentic Cloud - Agents. Models. Serverless.

Ship autonomous AI workers that plan, call tools (MCP), and deliver results - not just chat. Access any model through a unified API, or deploy your own custom models on serverless GPUs. One platform, pay per use, production-ready in minutes.

• Agents: build autonomous workers that plan, execute tools, and return structured results - then freeze a good run into a reusable, deterministic instruction you call as an API.
• Models: run any model through one unified API (image, video, audio, LLMs) in sync or async mode.
• Serverless: deploy your own custom models on autoscaling GPUs.
• MCP tools: plug in any MCP - web search, browsers, files, external APIs.

Agents tie everything together; Models and Serverless also work standalone - pick what you need and plug the rest in later.

Website

For more information, visit FlyMy.AI, read the docs, or join us on Discord.

Getting Started

This is the Python client for FlyMy.AI. Build and run agents, call any model, and drive serverless endpoints from Python - in sync or async mode.

Requirements

• Python 3.8+

Installation

Install the FlyMyAI client using pip:

pip install flymyai

Authentication

Before using the client, you need to have your API key, username, and project name. In order to get credentials, you have to sign up on flymy.ai and get your personal data on the profile.

📚 Core documentation: docs.flymy.ai - full guides for agents, inference, and MCP tools.

Agents

Autonomous agents plan, call tools (MCP), and return structured results. Declare an input_schema to make an agent reusable with runtime {{ variables }}, then freeze a good run into a Markdown instruction you can re-run as an API.

import asyncio
from flymyai import AsyncAgentClient

async def main():
    async with AsyncAgentClient(api_key="fly-secret-key") as client:
        # 1. Attach a tool (browse the full catalog with client.tools.available())
        tool = await client.tools.create(mcp_tool="tavily")

        # 2. Create a reusable agent. {{ variables }} require an input_schema.
        agent = await client.agents.create(
            name="News Brief",
            goal="Find the biggest news about {{ topic }} on {{ date }}. Return a one-line headline.",
            tools=[tool.id],
            input_schema={
                "type": "object",
                "properties": {"topic": {"type": "string"}, "date": {"type": "string"}},
                "required": ["topic", "date"],
            },
        )

        # 3. Run with variables; stream progress; get the structured result
        run = await client.runs.create(
            agent_id=agent.id,
            idempotency_key="news-brief-tesla-2026-05-21-v1",
            variables={"topic": "Tesla", "date": "2026-05-21"},
        )
        async for event in client.runs.stream_events(run.id):
            print(f"[{event.type}] {event.message}")
        result = await client.runs.wait(run.id)
        print(result.output)

        # 4. Chat: append a follow-up message and continue the same run
        await client.runs.append_message(run.id, text="Make it punchier.")
        await client.runs.wait(run.id)

        # 5. Freeze into a reusable instruction, re-run with fresh variables - your API
        compilation = await client.agents.compile_from_run(run.id)
        later = await client.compilations.run_instruction_and_wait(
            compilation.id,
            idempotency_key="news-brief-bitcoin-2026-05-19-v1",
            variables={"topic": "Bitcoin", "date": "2026-05-19"},
        )
        print(later.output)

asyncio.run(main())

Other agent methods: client.tools.available() / provide_config() / call(), client.runs.get() / list() / cancel(), client.agents.update() / suggest_schema(), client.compilations.update() (edit a frozen instruction). A synchronous AgentClient with the same method names (no await) is also available. Full reference: docs.flymy.ai/agents.

Limits for automatic subagents

Ordinary client.agents.run(...) calls use the same delegation runtime as chat. Pass subagent_limits={"cap_usd": "3", "max_children": 6, "max_parallel": 3} alongside the required idempotency_key to pin owner limits for a new run. cap_usd=0 disables helpers. The cap covers subagents and their tools; lead charges remain separate. The server validates limits before starting work and rejects a changed request under a reused idempotency key.

Sync and async clients support this option. Owner client.compilations.run_instruction(...) and run_instruction_and_wait(...) accept it too; omission inherits the frozen source run's limits. Scheduled runs inherit those limits automatically. Embedded customer calls cannot override owner limits. Owner run responses expose the effective subagent_limits.

Personal connection first

A new user starts in one implicit personal space. The first connection of a toolkit is the obvious default connection. Give that direct connection to the same agent and run it without naming projects, groups, slots, principals, revisions, or mappings.

import time
import webbrowser

from flymyai import AgentClient, McpAccessMode

with AgentClient(api_key="fly-secret-key") as client:
    # Omitting alias keeps the first personal Gmail connection on `default`.
    gmail = client.tools.create(mcp_tool="gmail")

    # Gmail setup is hosted. Open the returned URL, finish OAuth, then wait for
    # the same connection row to become ready before assigning it or running.
    if not gmail.is_configured:
        if gmail.redirect_url:
            webbrowser.open(gmail.redirect_url)
            input("Finish Gmail authorization, then press Enter: ")
        elif gmail.next_configuration_step:
            raise RuntimeError(
                "This connector needs an answer. Inspect "
                "gmail.next_configuration_step and call "
                "client.tools.provide_config(gmail.id, user_response=...)."
            )
        deadline = time.monotonic() + 300
        while time.monotonic() < deadline:
            gmail = client.tools.get(gmail.id)
            if gmail.is_configured:
                break
            time.sleep(2)
        else:
            raise TimeoutError("Gmail authorization did not become ready in 5 minutes.")

    agent = client.agents.create(
        name="Personal inbox helper",
        goal="Summarize my unread mail.",
        tools=[gmail.id],
        mcp_access_mode=McpAccessMode.LEGACY,
    )
    run = client.agents.run(
        agent.id,
        idempotency_key="personal-inbox-summary-v1",
    )

legacy here is the explicit compatibility projection for direct personal connections. The response also exposes agent.mcp_access_mode; old server responses that omit it parse as McpAccessMode.LEGACY. This simple path uses the same exact connection row as the controls below, so it can grow without rebuilding the agent or reconnecting the account.

Advanced connection access

Open the advanced model only when the same agent needs a second account of one toolkit, reusable sharing, group access, separate clients or environments, or explicit policy and mapping control.

from flymyai import (
    AgentClient,
    McpAccessMode,
    McpResourceSetManagementMode,
    McpResourceSetMemberInput,
    McpResourceType,
)

with AgentClient(api_key="fly-secret-key") as client:
    support = client.tools.create(mcp_tool="gmail", alias="support_mail")
    sales = client.tools.create(mcp_tool="gmail", alias="sales_mail")

    resource_set = client.mcp_resource_sets.create(
        name="Mail operations",
        management_mode=McpResourceSetManagementMode.FLYMYAI,
    )
    resource_set = client.mcp_resource_sets.replace_members(
        resource_set.id,
        expected_revision=resource_set.revision,
        members=[
            McpResourceSetMemberInput(
                resource_type=McpResourceType.USER_MCP_TOOL,
                resource_id=support.public_id,
                slot="support_read",
                allowed_actions=["GMAIL_SEARCH_EMAILS"],
            ),
            McpResourceSetMemberInput(
                resource_type=McpResourceType.USER_MCP_TOOL,
                resource_id=support.public_id,
                slot="support_send",
                allowed_actions=["GMAIL_SEND_EMAIL"],
            ),
            McpResourceSetMemberInput(
                resource_type=McpResourceType.USER_MCP_TOOL,
                resource_id=sales.public_id,
                slot="sales_mailbox",
            ),
        ],
    )
    agent = client.agents.create(
        name="Mail operations",
        goal="Work in the mailbox selected for this request.",
        mcp_resource_set_ids=[resource_set.id],
        mcp_access_mode=McpAccessMode.SCOPED,
    )

    # Metadata writes use the revision that this editor originally loaded.
    resource_set = client.mcp_resource_sets.update(
        resource_set.id,
        expected_revision=resource_set.revision,
        description="Support and sales mailboxes",
    )

Aliases and slots accept 1 to 64 and 1 to 128 characters respectively, using only ASCII letters, digits, underscores, and hyphens. They are labels, not authority. Select resources with stable public UUIDs. client.tools.list() follows bounded cursor pages and can filter by exact mcp_tool or alias. Metadata update() uses PATCH, complete metadata replace() uses PUT, and member replacement uses replace_members(); all three require expected_revision.

client.mcp_resource_sets.list() and client.agent_groups.list() follow the backend's compact cursor envelope across pages. They reject malformed or repeated cursors and stop at fixed page and row safety limits. A bounded plain array remains accepted for compatibility with an older server during rollout. Resource-set list rows are McpResourceSetSummary values with member_count and no nested members. Use client.mcp_resource_sets.get(id) for one bounded full set or client.mcp_resource_sets.list_members(id) to traverse a large membership collection. Sync and async clients expose the same contract.

Member identity is the exact (resource_type, resource_id, slot) tuple. The same resource UUID may therefore appear in several different slots, as the support connection does above, and every slot keeps its own allowed_actions ceiling. Repeating the same tuple is invalid. Members pooled inside one slot must use the same action ceiling.

Returning a scoped agent to personal legacy access is intentionally a safe multi-request workflow. First remove the agent from every group while preserving each group's other agents and resource sets, then clear its direct resource-set grants. Only after every cleanup request succeeds, switch the mode in a separate PATCH:

for group in client.agent_groups.list():
    if agent.id in group.agent_ids:
        client.agent_groups.replace_assignments(
            group.id,
            agent_ids=[item for item in group.agent_ids if item != agent.id],
            resource_set_ids=group.resource_set_ids,
        )

client.agents.update(agent.id, mcp_resource_set_ids=[])
client.agents.update(agent.id, mcp_access_mode=McpAccessMode.LEGACY)

If any cleanup request fails, do not send the final mode PATCH. The agent then remains fail-closed in scoped mode. The SDK deliberately does not hide these independent writes in a convenience helper: automatic rollback could restore stale grants or overwrite another editor's group assignment.

An owner project uses management_mode=McpResourceSetManagementMode.FLYMYAI and omits principal_id. A customer-managed named mapping uses the exact typed principal UUID. The complete embedded flow below obtains that principal with access.principal_for_external_user(customer_id), creates the mapping with principal_id=principal.id, installs exact connection UUID members, and runs with the returned mapping revision.

The deprecated external_principal_id field is not an accepted SDK argument. For a FlyMyAI-managed deployment run, omit both resource_set_id and connections so saved bindings apply. Send resource_set_id only for a customer-managed named mapping, or connections for a one-off explicit mapping. The last two are mutually exclusive.

AgentClient and AsyncAgentClient call the Agents REST API. They expose Python methods such as client.agents.create(), client.mcp_resource_sets.replace_members(), and client.deployments.run(). The Agents MCP gateway is a separate assistant surface with its own discovered tool names. Do not translate a REST SDK method into an invented MCP-only lifecycle call, and do not treat an MCP tool name as a Python method.

Embedded customer agents

An embedded deployment publishes one immutable frozen version while each customer authorizes their own MCP accounts. Hosted authorization links must be created by your trusted application backend because the FlyMyAI API key must never be sent to the customer browser.

client.deployments.access() is read-only and never creates a principal. It can inspect the published requirements. The first mutating customer bootstrap is client.deployments.create_connection_link(), which calls REST POST /api/v1/agents/deployments/{deployment_id}/connect-session/ with the stable external_user_id and one exact required slot.

Complete synchronous flow for a customer-managed named mapping:

import os
import time
import webbrowser

from flymyai import (
    AgentClient,
    McpResourceSetManagementMode,
    McpResourceSetMemberInput,
    McpResourceType,
)

customer_id = os.environ["PRODUCT_CUSTOMER_ID"]  # Stable, non-secret product ID
agent_id = os.environ["FLYMYAI_AGENT_ID"]
accepted_run_id = int(os.environ["FLYMYAI_ACCEPTED_RUN_ID"])

with AgentClient(api_key="fly-secret-key") as client:
    # compile_from_run freezes and polls until compilation has finished.
    compilation = client.agents.compile_from_run(accepted_run_id)
    version = next(
        item
        for item in client.versions.list(agent_id=agent_id)
        if item.source_compilation == compilation.id
    )

    deployment = client.deployments.create(
        agent_id=agent_id,
        version_id=version.id,
        name="Production",
    )
    manifest = client.deployments.access(deployment.id)  # Read-only.
    preflight = client.deployments.preflight(
        deployment.id,
        publish_mode="embedded",
        version_id=version.id,
    )
    if not preflight.ready:
        raise RuntimeError("Deployment preflight is not ready.")
    deployment = client.deployments.publish(
        deployment.id,
        publish_mode="embedded",
        version_id=version.id,
    )

    # Each POST is the mutating principal/bootstrap step for one exact slot.
    sessions = [
        client.deployments.create_connection_link(
            deployment.id,
            external_user_id=customer_id,
            slot=requirement.slot,
        )
        for requirement in manifest.requirements
        if requirement.connection_required
    ]
    for session in sessions:
        webbrowser.open(session.redirect_url)
    if sessions:
        input("Finish every customer authorization, then press Enter: ")

    # Poll the filtered access projection. The typed helpers fail closed until
    # one exact principal and active connection IDs exist for every slot.
    deadline = time.monotonic() + 300
    while True:
        access = client.deployments.access(
            deployment.id,
            external_user_id=customer_id,
        )
        try:
            principal = access.principal_for_external_user(customer_id)
            ready_by_slot = {
                requirement.slot: access.ready_connection_ids_for_slot(
                    principal_id=principal.id,
                    slot=requirement.slot,
                )
                for requirement in access.requirements
                if requirement.connection_required
            }
            break
        except ValueError:
            if time.monotonic() >= deadline:
                raise TimeoutError("Customer connections did not become ready.")
            time.sleep(2)

    customer_mapping = client.mcp_resource_sets.create(
        name=f"{customer_id} connections",
        management_mode=McpResourceSetManagementMode.CUSTOMER,
        principal_id=principal.id,
    )
    customer_mapping = client.mcp_resource_sets.replace_members(
        customer_mapping.id,
        expected_revision=customer_mapping.revision,
        members=[
            McpResourceSetMemberInput(
                resource_type=McpResourceType.INTEGRATION_CONNECTION,
                resource_id=connection_id,
                slot=requirement.slot,
                allowed_actions=requirement.exact_actions,
            )
            for requirement in access.requirements
            if requirement.connection_required
            for connection_id in ready_by_slot[requirement.slot]
        ],
    )
    result = client.deployments.run_and_wait(
        deployment.id,
        variables={"topic": "Q3 pipeline"},
        external_user_id=customer_id,
        resource_set_id=customer_mapping.id,
        resource_set_revision=customer_mapping.revision,
        idempotency_key=f"{customer_id}-q3-pipeline-v1",
    )

AsyncAgentClient exposes the same typed lifecycle. Await the network methods; principal_for_external_user() and ready_connection_ids_for_slot() remain ordinary local model helpers.

The access request must include the exact external_user_id; do not select a principal or connection by display name, email, alias, provider account ID, or row order. McpResourceSets.list() supports query, authority_type, and principal_id; AgentGroups.list() supports query. These filters remain on every compact-cursor page.

client.deployments.run() and run_and_wait() call the stable deployment endpoint, so callers need the deployment UUID but never need a compilation ID. Normally a run uses the saved customer bindings. To choose among several authorized accounts for one run, pass connections={"sender_inbox": "8335876a-ee78-45db-9d49-0ae148bd0158"} or a list of up to 25 connection UUIDs for a multi-connection slot. Provider account IDs and aliases are not accepted in this mapping. Customer file attachments are not enabled for embedded runs in this beta. Wait for every hosted authorization callback to complete before starting the customer's first run.

The legacy client.compilations.run_instruction() and run_instruction_and_wait() methods remain available for compilation-scoped owner runs. Deprecated client.compilations.run() and its async counterpart fail locally before HTTP because that legacy endpoint has no caller-owned replay contract.

Every agents.run(), runs.create(), tools.call(), compilations.run_instruction(), and deployments.run() call, including their wait helpers and async variants, requires a caller-owned idempotency_key. It must be non-blank and no longer than 255 characters and contain no control or non-printable characters. The SDK forwards the exact value as Idempotency-Key and never generates one for the caller. Reuse a key only for an identical retry; choose a new key for a new logical execution. This is an intentional compatibility break for keyless effect calls because a lost response must not create a second execution that repeats external writes.

Neural Network Inference

Run any model on the platform with flymyai.async_run (async) or flymyai.run (sync).

Image generation - Nano Banana 🍌

import asyncio
import base64
import flymyai

async def main():
    response = await flymyai.async_run(
        apikey="fly-secret-key",
        model="flymyai/nano-banana",
        payload={"prompt": "a cute cat astronaut floating in a neon nebula, studio lighting"},
    )
    with open("nano_banana.jpg", "wb") as f:
        f.write(base64.b64decode(response.output_data["image"][0]))

asyncio.run(main())

Video generation - Veo 3.1 Fast

import asyncio
import flymyai

async def main():
    response = await flymyai.async_run(
        apikey="fly-secret-key",
        model="flymyai/veo31-fast-generate",
        payload={"prompt": "a red sports car driving along a coastal road at sunset, cinematic"},
    )
    print(response.output_data["video"][0])  # public URL to the generated .mp4

asyncio.run(main())

Parallel generation

Fire many requests concurrently with asyncio.gather:

import asyncio
import base64
import flymyai

PROMPTS = ["a neon city at night", "a serene mountain lake at dawn", "a retro robot barista"]

async def main():
    results = await asyncio.gather(*[
        flymyai.async_run(
            apikey="fly-secret-key",
            model="flymyai/nano-banana",
            payload={"prompt": p},
        )
        for p in PROMPTS
    ])
    for i, r in enumerate(results):
        with open(f"img_{i}.jpg", "wb") as f:
            f.write(base64.b64decode(r.output_data["image"][0]))

asyncio.run(main())

Custom MCP servers and OAuth

client.mcp_servers adds an MCP server by URL - a service that is not in the catalog, or a provider's own MCP endpoint - and calls it. auth_type is none, api_key, bearer_token, basic or oauth. With oauth FlyMyAI discovers the server's authorization and registers itself where the provider allows it; authorize returns a link a person opens to approve, and then the server is connected.

import webbrowser

from flymyai import AgentClient
from flymyai.agents import McpServerOAuthError

with AgentClient(api_key="fly-secret-key") as client:
    server = client.mcp_servers.create(
        name="Linear", url="https://mcp.linear.app/mcp", auth_type="oauth"
    )
    try:
        authorization = client.mcp_servers.authorize(server.id)
    except McpServerOAuthError as error:
        if error.code != "oauth_client_required":
            raise
        # No automatic registration (Slack's MCP is like this): register
        # error.redirect_uri in your own OAuth app, save its client ID and
        # secret on the server, then authorize again.
        client.mcp_servers.update(
            server.id, oauth_client_id="...", oauth_client_secret="..."
        )
        authorization = client.mcp_servers.authorize(server.id)
    webbrowser.open(authorization.authorize_url)  # works once, 15 minutes

    # After the approval: status "connected", the tools in discovered_tools.
    server = client.mcp_servers.get(server.id)
    if server.status == "connected":
        client.mcp_servers.call(
            server.id, "list_issues", {"first": 5}, idempotency_key="issues-1"
        )

server.oauth (owner only) says authorized, reconnect_required (authorize again), issuer, scopes, client and redirect_uri; tokens and the client secret never come back. disconnect forgets the grant.

Personal MCP sharing

The optional AgentClient.shares and AsyncAgentClient.shares clients manage email invitations, exact-connection grants, device tokens and revocation. A recipient must verify the addressed email and explicitly pass accept_billing=True when accepting. FlyMy execution charges use the recipient's wallet, without owner-wallet fallback. Provider charges remain with the connected account; existing team MCP calls continue using the team wallet.

Personal sharing is live on FlyMy.AI production (backend and Agents MCP) since 2026-09-25; see the sharing guide. Self-hosted deployments keep it behind the independent MCP_PERSONAL_SHARING_ENABLED gate, off by default. Existing inference clients and resource-set/agent contracts remain available.

Frontend artifacts

AgentClient.artifacts and AsyncAgentClient.artifacts manage frontend artifacts (flymy.artifact.v1): a small web page, a presentation or a mini game with immutable versions. Share it by link (with or without its sources) or with people by name to view or edit, and clone anyone's shared artifact into your own copy. Artifacts are on for every account; client.artifacts.status() returns your limits.

from flymyai import AgentClient
from flymyai.agents import artifact_file, artifact_files_from_directory

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

# a folder (a built site, a game, a page copied from a Claude artifact)
made = client.artifacts.create(
    name="Pod racer",
    files=artifact_files_from_directory("./pod-racer"),
    visibility="link",
    idempotency_key="pod-racer-create-1",
)
racer = made.artifact
print(racer.share_url)

# a new version on top of the latest; the other files are kept
client.artifacts.publish(
    racer.id,
    base_version=racer.latest_version,
    files=[artifact_file("js/app.js", "speed = 2")],
    message="Faster pods",
    idempotency_key="pod-racer-v2",
)

# invite a teammate to publish versions too ("view" lets them read and clone);
# they accept first, and nothing is shared with them until then
invited = client.artifacts.add_member(racer.id, "teammate@example.com", role="edit")
print(invited.member.status)  # pending

# on the teammate's side: what waits for them, and their answer
for invitation in client.artifacts.invitations().results:
    client.artifacts.accept_invitation(invitation.id)
shared_with_me = client.artifacts.list(scope="shared")

# your own copy of someone's artifact shared with sources
copy = client.artifacts.clone(
    share_link="https://app.flymy.ai/artifacts/s/<handle>",
    idempotency_key="clone-1",
)

Sharing by name never tells anyone who an email or a username belongs to: an email answers the same whether an account has it, the owner sees each person as they typed them (invited_as) with status pending or accepted, and a member sees only which row is theirs (you). remove_member(artifact_id, member.id) stops sharing with an invited or accepted person; a member removes their own row to leave.

Writes that create something take a caller-owned idempotency_key: reuse it only to retry the identical call. A version published on a stale base_version raises ArtifactStaleBaseVersionError with latest_version. These are the v1 methods and they stay: a breaking change would arrive as new methods next to them, never in place.

Projects

AgentClient.projects and AsyncAgentClient.projects read and run your projects (flymy.project.v1): an app applied from one flymy.yaml (its pages, agents, servers, storage and budgets), a page published on its own, a fleet (a lead agent and the agents it starts) or a frontend artifact. A project is its config: creating, stopping and starting it are a plan and an apply of that config, and each leaves a release.

from flymyai import AgentClient

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

for project in client.projects.list().projects:
    print(project.id, project.status)

# a plan creates nothing: show its price, then create exactly that plan
plan = client.projects.plan(name="paint-arena", budget={"per_day_usd": "5"})
print(plan.plan["usd_per_hour"], plan.plan["changes"])
made = client.projects.create(
    name="paint-arena",
    budget={"per_day_usd": "5"},
    plan_token=plan.plan_token,
    idempotency_key="paint-arena-create-1",
)

# stop it (its servers stop and the rest of their holds is refunded), start it again
client.projects.stop("app:me/paint-arena", idempotency_key="paint-arena-stop-1")
start = client.projects.start("app:me/paint-arena")  # the plan, nothing changes
print(start.plan["usd_per_hour"])
client.projects.start(
    "app:me/paint-arena",
    plan_token=start.plan_token,
    idempotency_key="paint-arena-start-1",
)

journal = client.projects.errors("app:me/paint-arena", limit=20)

Change an app by applying its flymy.yaml again with AgentClient.apps: plan shows what it would create, change or stop and its price, and apply applies exactly that plan as a release.

import time

from flymyai.agents import artifact_files_from_directory

files = artifact_files_from_directory("./paint-arena")  # flymy.yaml at the root
plan = client.apps.plan(files=files)
print(plan.changes, plan.usd_per_hour)  # show the user before applying
release = client.apps.apply(
    files=files, plan_token=plan.plan_token, idempotency_key="paint-arena-apply-2"
)
while client.apps.status(release.release).status == "applying":
    time.sleep(5)

client.apps.files("me/paint-arena") reads the applied template back; plan and apply also take app="me/paint-arena" with overrides to replace some files and keep the rest. get, errors, templates and agent (the project's own agent, created on first use) complete the projects set. A fleet (fleet:<lead agent id>) and a frontend artifact (artifact:<artifact id>) are read with get and errors; they have no project agent and are not stopped or started (an artifact changes through client.artifacts).

Advanced agent helpers

Draft an input_schema from a prompt

import asyncio
from flymyai import AsyncAgentClient

async def main():
    async with AsyncAgentClient(api_key="fly-secret-key") as client:
        suggestion = await client.agents.suggest_schema(
            user_prompt="Summarize {{ url }} in {{ n_sentences }} sentences.",
            generate_descriptions=True,
        )
        print(suggestion.input_schema)
        print(suggestion.input_description)

asyncio.run(main())

Handle invalid variables

When variables don't match the agent's input_schema, the server returns HTTP 400 and the client raises VariablesValidationError:

from flymyai import VariablesValidationError

try:
    await client.agents.run(
        agent.id,
        idempotency_key="validate-agent-input-v1",
        variables={},
    )
except VariablesValidationError as err:
    print(err.messages)      # ["'url' is a required property", ...]
    print(err.field_errors)  # {"url": "'url' is a required property"}

Draft schemas from a finished run

# Infer schemas from a completed run's chat + tool trace
# (also persists them onto the source agent).
suggestion = await client.runs.suggest_schema(
    run.id,
    inputs_prompt="A URL and a sentence count",
    outputs_prompt="A short summary",
)