flyte-sdk-agent

Flyte 2 SDK Agent Skill

Build durable, observable AI agents with Flyte 2.

Grounding References

Resource	URL
Official docs	https://www.union.ai/docs/v2/flyte
Docs index (LLMs)	https://www.union.ai/docs/v2/flyte/llms.txt
SDK API reference	https://www.union.ai/docs/v2/union/api-reference/flyte-sdk/
CLI API reference	https://www.union.ai/docs/v2/union/api-reference/flyte-cli/
flyte-sdk source	https://github.com/flyteorg/flyte-sdk
Example code	https://github.com/unionai/unionai-examples
Flyte MCP tools	Available via the flyte-cluster and flyte-docs MCP servers

Ground unfamiliar APIs in real examples. When unsure of a current Flyte 2 API, or for a pattern not shown below, and the flyte-docs search tools are available, search them first — by exact symbol (TaskEnvironment, flyte.io.File, map_task), since matching is literal substring, not semantic — then adapt a real example rather than inventing one, and cite the file or section you pulled it from. (Flyte 2 is not flytekit; priors are often wrong.)

Pure Python Agents (No Framework)

ReAct Pattern — Reason, Act, Observe

import flyte

env = flyte.TaskEnvironment(
    name="react-agent",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "openai", "boto3",
    ),
)

@env.task
async def think(observation: str) -> str:
    """LLM generates next action."""
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": f"Observation: {observation}\nWhat do you do next?"},
        ],
    )
    return response.choices[0].message.content

@env.task
async def act(thought: str) -> dict:
    """Parse thought and execute tool call."""
    # Parse the thought to extract tool name and arguments
    # Then call the appropriate tool
    return {"tool": "search", "result": "search results..."}

@env.task
async def observe(result: dict) -> str:
    """Format tool result for the next reasoning step."""
    return f"Tool {result['tool']} returned: {result['result']}"

@env.task
async def react_loop(initial_query: str, max_steps: int = 5) -> str:
    """ReAct loop: think → act → observe → think → ..."""
    observation = initial_query
    for i in range(max_steps):
        thought = await think(observation)
        if "FINAL_ANSWER" in thought:
            return thought.split("FINAL_ANSWER:")[-1].strip()
        result = await act(thought)
        observation = await observe(result)
    return "Max steps reached"

if __name__ == "__main__":
    import asyncio
    result = asyncio.run(react_loop("What is the weather in Tokyo?"))
    print(result)