Update to Flyte

{
  "description": "Builds durable agents with Flyte 2 — ReAct patterns, Plan-and-Execute, LangGraph integration, PydanticAI integration, OpenAI Agents SDK integration, agent memory, MCP tool integration, skills, and agent chat UI. Use when the user wants to build AI agents, implement ReAct loops, integrate agent frameworks, add agent memory, or build agent-powered workflows. Trigger words: \"agent\", \"ReAct\", \"LangGraph\", \"PydanticAI\", \"OpenAI agents\", \"MCP\", \"tool calling\", \"memory\", \"skills\", \"agentic\".",
  "included_files": [],
  "name": "flyte-sdk-agent",
  "skill_md_contents": "---\nname: flyte-sdk-agent\ndescription: 'Builds durable agents with Flyte 2 — ReAct patterns, Plan-and-Execute, LangGraph integration, PydanticAI integration, OpenAI Agents SDK integration, agent memory, MCP tool integration, skills, and agent chat UI. Use when the user wants to build AI agents, implement ReAct loops, integrate agent frameworks, add agent memory, or build agent-powered workflows. Trigger words: \"agent\", \"ReAct\", \"LangGraph\", \"PydanticAI\", \"OpenAI agents\", \"MCP\", \"tool calling\", \"memory\", \"skills\", \"agentic\".'\n---\n\n# Flyte 2 SDK Agent Skill\n\nBuild durable, observable AI agents with Flyte 2.\n\n## Grounding References\n\n| Resource | URL |\n|---|---|\n| Official docs | https://www.union.ai/docs/v2/flyte |\n| Docs index (LLMs) | https://www.union.ai/docs/v2/flyte/llms.txt |\n| SDK API reference | https://www.union.ai/docs/v2/union/api-reference/flyte-sdk/ |\n| CLI API reference | https://www.union.ai/docs/v2/union/api-reference/flyte-cli/ |\n| flyte-sdk source | https://github.com/flyteorg/flyte-sdk |\n| Example code | https://github.com/unionai/unionai-examples |\n| Flyte MCP tools | Available via the `flyte-cluster` and `flyte-docs` MCP servers |\n\n**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.)\n\n## Pure Python Agents (No Framework)\n\n### ReAct Pattern — Reason, Act, Observe\n\n```python\nimport flyte\n\nenv = flyte.TaskEnvironment(\n    name=\"react-agent\",\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"openai\", \"boto3\",\n    ),\n)\n\n@env.task\nasync def think(observation: str) -> str:\n    \"\"\"LLM generates next action.\"\"\"\n    from openai import OpenAI\n    client = OpenAI()\n    response = client.chat.completions.create(\n        model=\"gpt-4\",\n        messages=[\n            {\"role\": \"system\", \"content\": \"You are a helpful assistant.\"},\n            {\"role\": \"user\", \"content\": f\"Observation: {observation}\\nWhat do you do next?\"},\n        ],\n    )\n    return response.choices[0].message.content\n\n@env.task\nasync def act(thought: str) -> dict:\n    \"\"\"Parse thought and execute tool call.\"\"\"\n    # Parse the thought to extract tool name and arguments\n    # Then call the appropriate tool\n    return {\"tool\": \"search\", \"result\": \"search results...\"}\n\n@env.task\nasync def observe(result: dict) -> str:\n    \"\"\"Format tool result for the next reasoning step.\"\"\"\n    return f\"Tool {result['tool']} returned: {result['result']}\"\n\n@env.task\nasync def react_loop(initial_query: str, max_steps: int = 5) -> str:\n    \"\"\"ReAct loop: think → act → observe → think → ...\"\"\"\n    observation = initial_query\n    for i in range(max_steps):\n        thought = await think(observation)\n        if \"FINAL_ANSWER\" in thought:\n            return thought.split(\"FINAL_ANSWER:\")[-1].strip()\n        result = await act(thought)\n        observation = await observe(result)\n    return \"Max steps reached\"\n\nif __name__ == \"__main__\":\n    import asyncio\n    result = asyncio.run(react_loop(\"What is the weather in Tokyo?\"))\n    print(result)\n```\n\n### Plan-and-Execute with Parallel Fan-out\n\n```python\n@env.task\nasync def plan(query: str) -> list[str]:\n    \"\"\"Generate a plan of sub-tasks.\"\"\"\n    from openai import OpenAI\n    client = OpenAI()\n    response = client.chat.completions.create(\n        model=\"gpt-4\",\n        messages=[\n            {\"role\": \"system\", \"content\": \"Break this query into sub-tasks.\"},\n            {\"role\": \"user\", \"content\": query},\n        ],\n    )\n    # Parse response into list of sub-tasks\n    return response.choices[0].message.content.split(\"\\n\")\n\n@env.task\nasync def execute_subtask(task: str) -> str:\n    \"\"\"Execute a single sub-task.\"\"\"\n    ...\n    return result\n\n@env.task\nasync def synthesize(results: list[str]) -> str:\n    \"\"\"Synthesize results into a final answer.\"\"\"\n    from openai import OpenAI\n    client = OpenAI()\n    response = client.chat.completions.create(\n        model=\"gpt-4\",\n        messages=[\n            {\"role\": \"system\", \"content\": \"Synthesize these results.\"},\n            {\"role\": \"user\", \"content\": \"\\n\".join(results)},\n        ],\n    )\n    return response.choices[0].message.content\n\n@env.task\nasync def plan_and_execute(query: str) -> str:\n    \"\"\"Plan sub-tasks, execute in parallel, synthesize results.\"\"\"\n    tasks = await plan(query)\n    # Parallel execution of sub-tasks\n    results = await flyte.map(execute_subtask, tasks)\n    return await synthesize(results)\n```\n\n## Agent Framework Integrations\n\n### LangGraph Agents\n\n```python\nimport flyte\nfrom langgraph.prebuilt import create_react_agent\nfrom langchain_openai import ChatOpenAI\n\nenv = flyte.TaskEnvironment(\n    name=\"langgraph-agent\",\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"langgraph\", \"langchain-openai\", \"langchain\",\n    ),\n)\n\n@env.task\nasync def langgraph_agent(query: str) -> str:\n    \"\"\"Run a LangGraph agent as a Flyte task.\"\"\"\n    agent = create_react_agent(\n        model=ChatOpenAI(model=\"gpt-4\"),\n        tools=[search_tool, calculate_tool],\n    )\n    result = agent.invoke({\"messages\": [(\"user\", query)]})\n    return result[\"messages\"][-1].content\n\n@env.task\nasync def parallel_agents(queries: list[str]) -> list[str]:\n    \"\"\"Run multiple LangGraph agents in parallel.\"\"\"\n    results = await flyte.map(langgraph_agent, queries)\n    return results\n```\n\n### PydanticAI Agents\n\n```python\nimport flyte\nfrom pydantic_ai import Agent, RunContext\nfrom pydantic import BaseModel\n\nclass AnalysisResult(BaseModel):\n    summary: str\n    confidence: float\n    recommendations: list[str]\n\nenv = flyte.TaskEnvironment(\n    name=\"pydantic-agent\",\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"pydantic-ai\", \"openai\",\n    ),\n)\n\n@env.task\nasync def pydantic_agent(query: str) -> AnalysisResult:\n    \"\"\"Run a PydanticAI agent with structured output.\"\"\"\n    agent = Agent(\n        \"openai:gpt-4\",\n        result_type=AnalysisResult,\n    )\n    result = await agent.run(query)\n    return result.data\n\n@env.task\nasync def parallel_pydantic_agents(queries: list[str]) -> list:\n    \"\"\"Run multiple PydanticAI agents in parallel.\"\"\"\n    results = await flyte.map(pydantic_agent, queries)\n    return results\n```\n\n### OpenAI Agents SDK\n\n```python\nimport flyte\nfrom openai.agents import Agent, Tool\n\nenv = flyte.TaskEnvironment(\n    name=\"openai-agent\",\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"openai\",\n    ),\n)\n\n@env.task\nasync def search_tool(query: str) -> str:\n    \"\"\"A tool that is also a durable Flyte task.\"\"\"\n    # Tools become durable — results are cached and replayable\n    ...\n    return results\n\n@env.task\nasync def openai_agent(query: str) -> str:\n    \"\"\"Run an OpenAI Agents SDK agent.\"\"\"\n    agent = Agent(\n        name=\"researcher\",\n        instructions=\"Research the query thoroughly.\",\n        tools=[search_tool],\n    )\n    result = agent.run(query)\n    return result.final_output\n\n@env.task\nasync def main(query: str) -> str:\n    \"\"\"Wrap the agent in a Flyte workflow for durability.\"\"\"\n    return await openai_agent(query)\n```\n\n## Building Agents with Flyte Primitives\n\n### How Flyte maps to the agent stack\n\n| Agent Concept | Flyte Primitive |\n|---|---|\n| Tool call | `@env.task` — each tool is a durable task |\n| Reasoning step | `@env.task` with LLM call |\n| Observation | Output of tool task → input to reasoning task |\n| Loop | `flyte.new_condition` for external gates, or dynamic workflows for internal loops |\n| Parallel execution | `flyte.map` for fan-out sub-tasks |\n| Traces | `flyte.trace` for lightweight LLM calls within a step |\n\n### Durable agent pattern\n\n```python\n@env.task\nasync def llm_call(prompt: str, model: str = \"gpt-4\") -> str:\n    \"\"\"Durable LLM call — cached by input, replayable.\"\"\"\n    from openai import OpenAI\n    client = OpenAI()\n    response = client.chat.completions.create(model=model, messages=[{\"role\": \"user\", \"content\": prompt}])\n    return response.choices[0].message.content\n\n@env.task\nasync def tool_call(tool_name: str, args: dict) -> str:\n    \"\"\"Durable tool execution — each tool call is a Flyte action.\"\"\"\n    if tool_name == \"search\":\n        return perform_search(args[\"query\"])\n    elif tool_name == \"calc\":\n        return str(evaluate(args[\"expression\"]))\n    raise ValueError(f\"Unknown tool: {tool_name}\")\n\n@env.task\nasync def agent_step(\n    history: list[dict],\n) -> dict:\n    \"\"\"Single agent step: decide next action.\"\"\"\n    from openai import OpenAI\n    client = OpenAI()\n    response = client.chat.completions.create(\n        model=\"gpt-4\",\n        messages=[\n            {\"role\": \"system\", \"content\": \"Decide next action based on history.\"},\n            *history,\n        ],\n        response_format={\"type\": \"json_object\"},\n    )\n    return eval(response.choices[0].message.content)  # parse JSON\n\n@env.task\nasync def run_agent(\n    query: str,\n    max_steps: int = 10,\n) -> str:\n    \"\"\"Run a durable agent loop.\"\"\"\n    history = [{\"role\": \"user\", \"content\": query}]\n\n    for step in range(max_steps):\n        decision = await agent_step(history)\n        if decision[\"type\"] == \"final_answer\":\n            return decision[\"answer\"]\n\n        # Execute tool\n        result = await tool_call(decision[\"tool\"], decision[\"args\"])\n        history.append({\"role\": \"assistant\", \"content\": f\"Tool {decision['tool']} → {result}\"})\n\n    return \"Max steps reached\"\n```\n\n## Agent Memory\n\n### Keyed MemoryStore\n\n```python\nimport flyte\nfrom flyte.extend import MemoryStore\n\nenv = flyte.TaskEnvironment(\n    name=\"agent-with-memory\",\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"openai\",\n    ),\n)\n\n@env.task\nasync def agent_with_memory(query: str, user_id: str) -> str:\n    \"\"\"Agent with persistent memory per user.\"\"\"\n    memory = MemoryStore(keyed_by=user_id)\n\n    # Load previous context\n    context = await memory.get(\"conversation\", default=[])\n\n    # Add new message\n    context.append({\"role\": \"user\", \"content\": query})\n\n    # Generate response\n    from openai import OpenAI\n    client = OpenAI()\n    response = client.chat.completions.create(\n        model=\"gpt-4\",\n        messages=context,\n    )\n    answer = response.choices[0].message.content\n\n    # Store updated context\n    context.append({\"role\": \"assistant\", \"content\": answer})\n    await memory.set(\"conversation\", context)\n\n    return answer\n```\n\n### Run-level context\n\n```python\n@env.task\nasync def agent_step(query: str) -> str:\n    \"\"\"Access run-level context for memory.\"\"\"\n    ctx = flyte.ctx()\n\n    # Use run ID as a key for temporary memory\n    run_memory_key = f\"run:{ctx.run_id}:memory\"\n\n    # Store intermediate results\n    ...\n```\n\n## Agent Chat UI\n\n### Built-in chat UI\n\n```python\nimport flyte\nfrom flyte.extend import Agent, tool\n\nenv = flyte.TaskEnvironment(\n    name=\"chat-agent\",\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"openai\",\n    ),\n)\n\nclass ChatAgent(Agent):\n    \"\"\"Built-in agent with chat UI.\"\"\"\n\n    @tool\n    async def search(self, query: str) -> str:\n        \"\"\"Search the web.\"\"\"\n        ...\n\n    @tool\n    async def calculate(self, expression: str) -> str:\n        \"\"\"Evaluate a math expression.\"\"\"\n        ...\n\n    async def run(self, message: str) -> str:\n        \"\"\"Main agent loop.\"\"\"\n        # Use tools to respond\n        return \"Response\"\n\nif __name__ == \"__main__\":\n    flyte.init_from_config()\n    agent = ChatAgent()\n    agent.run(message=\"Hello!\")\n```\n\n### Custom FastAPI chat app\n\n```python\nfrom fastapi import FastAPI\nimport flyte\nfrom flyte.app.extras import FastAPIAppEnvironment\n\napp = FastAPI()\nenv = FastAPIAppEnvironment(\n    name=\"chat-app\",\n    app=app,\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"fastapi\", \"uvicorn\", \"openai\",\n    ),\n)\n\n@app.post(\"/chat\")\nasync def chat(message: str):\n    \"\"\"Chat endpoint — delegates to a Flyte agent task.\"\"\"\n    result = await flyte.run(agent_task, inputs={\"message\": message})\n    return {\"response\": result.outputs}\n\nif __name__ == \"__main__\":\n    flyte.init_from_config()\n    flyte.serve(env)\n```\n\n## Deploying Agents\n\n### As a task\n\n```bash\n# Run agent as a one-shot task\nflyte run agent.py run_agent --query \"Research X\"\n```\n\n### As a scheduled task (Trigger)\n\n```bash\n# Create a trigger for periodic agent execution\nflyte create trigger agent_task daily-agent \\\n  --schedule \"0 9 * * 1\"  # every Monday at 9am\n```\n\n### Behind a webhook\n\n```python\n# Agent behind a webhook app\n@app.post(\"/agent\")\nasync def agent_webhook(payload: dict):\n    flyte.run(agent_task, inputs={\"message\": payload[\"text\"]})\n    return {\"status\": \"queued\"}\n```\n\n### Chat app pattern\n\n```bash\n# Deploy as a persistent app\nflyte deploy agent_app.py env\n```\n\n## MCP Integration\n\n### Building an MCP server for agents\n\n```python\nimport flyte\nfrom flyte.extend import MCPServer, tool\n\nenv = flyte.TaskEnvironment(\n    name=\"mcp-server\",\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"fastmcp\",\n    ),\n)\n\n@env.task\nasync def build_mcp_server() -> MCPServer:\n    \"\"\"Build an MCP server with durable tools.\"\"\"\n\n    @tool\n    async def search(query: str) -> str:\n        return perform_search(query)\n\n    @tool\n    async def fetch(url: str) -> str:\n        return fetch_url(url)\n\n    return MCPServer(tools=[search, fetch])\n```\n\n### Connecting an MCP client\n\n```python\n# Claude Code — local (stdio)\n# Configure inClaude Code settings to connect to the Flyte MCP server\n\n# OpenCode — local\n# Configure in opencode.json to connect to the Flyte MCP server\n```\n\n## Agent Anti-Patterns\n\n1. **Don't put LLM calls in a loop without durability** — each LLM call should be a `@env.task` for caching and replay.\n2. **Don't use Union-only features** — avoid `ReusePolicy` and other Union-specific APIs.\n3. **Don't pass large prompts inline** — use `flyte.io.File` for large context windows.\n4. **Don't forget to set resources** — LLM agent tasks need CPU for the orchestration container.\n5. **Don't hardcode API keys** — use Flyte secrets for LLM API keys.\n"
}