The agent primitive

Flux has two agent surfaces. The agent() function from flux.tasks.ai creates a Flux @task backed by an LLM; you await it inside a @workflow and it participates in durable execution, retries, and the event log. The Agent harness is a separate surface: a standalone long-running process driven by a YAML file and the flux agent CLI. This page covers agent(), the in-workflow primitive.

Import and provider format

from flux.tasks.ai import agent

The model parameter takes a string in "provider/model_name" format:

If the string does not contain a /, Flux raises a ValueError before reaching the provider; the error message includes the valid formats.

Install the provider SDK you need:

pip install anthropic        # Anthropic Claude
pip install openai           # OpenAI
pip install google-genai     # Google Gemini
pip install ollama           # Ollama (local)
# or install all at once:
pip install flux-core[ai]

What agent() returns

agent() is an async factory. It returns a Flux @task; the response string comes from calling that task with an instruction:

assistant = await agent(
    "You are a helpful assistant.",
    model="anthropic/claude-sonnet-4-20250514",
)
response: str = await assistant("What is the capital of France?")

The returned task has this signature:

async def agent_task(instruction: str, *, context: str = "") -> str | BaseModel

• instruction — the user message sent to the LLM.
• context — optional background text prepended to the instruction. Use it to inject retrieved documents or earlier results without adding them to the conversation history.

The return type is str by default. When you pass response_format, it returns an instance of that Pydantic model instead.

Minimal Anthropic example

import os
from flux import ExecutionContext, workflow
from flux.tasks.ai import agent

@workflow
async def answer_question(ctx: ExecutionContext):
    assistant = await agent(
        "You are a concise assistant. Answer in one sentence.",
        model="anthropic/claude-sonnet-4-20250514",
    )
    return await assistant(ctx.input["question"])

Run it:

export ANTHROPIC_API_KEY="sk-ant-..."
flux workflow run answer_question '{"question": "Why is the sky blue?"}'

Same workflow with Ollama

Replace the model string. No API key required; start ollama serve first and pull the model:

ollama pull qwen3:8b

from flux import ExecutionContext, workflow
from flux.tasks.ai import agent

@workflow
async def answer_question_local(ctx: ExecutionContext):
    assistant = await agent(
        "You are a concise assistant. Answer in one sentence.",
        model="ollama/qwen3:8b",
    )
    return await assistant(ctx.input["question"])

The rest of the code is identical. Ollama is the recommended option for local development and offline iteration; see Choosing a provider for a full comparison.

Multi-turn conversations

Call the returned task multiple times within the same workflow to build a conversation loop. Add working_memory to carry conversation history across turns:

from flux import ExecutionContext, workflow
from flux.tasks.ai import agent
from flux.tasks.ai.memory import working_memory
from flux.tasks import pause

@workflow
async def chatbot(ctx: ExecutionContext):
    assistant = await agent(
        "You are a helpful assistant. Keep answers concise.",
        model="anthropic/claude-sonnet-4-20250514",
        working_memory=working_memory(),
    )

    response = await assistant(ctx.input["message"])
    print(response)

    for turn in range(10):
        resume_input = await pause(f"turn_{turn}")
        next_message = resume_input.get("message", "") if resume_input else ""
        if not next_message or next_message.lower() == "quit":
            break
        response = await assistant(next_message)
        print(response)

    return "Conversation ended."

Without working_memory, each await assistant(...) call is stateless. The LLM receives only the system prompt and the current instruction, with no memory of earlier turns.

Task options

Because agent() returns a Flux @task, you can attach task-level options with .with_options():

assistant = (await agent(
    "You are a research assistant.",
    model="anthropic/claude-sonnet-4-20250514",
)).with_options(
    retry_max_attempts=3,
    timeout=120,
)

This gives the agent task automatic retry on transient API errors and a 120-second deadline per invocation.

Full agent() signature

async def agent(
    system_prompt: str,
    *,
    model: str,
    name: str | None = None,
    description: str | None = None,
    tools: list[task] | None = None,
    skills: SkillCatalog | None = None,
    agents: list | None = None,
    planning: bool = False,
    max_plan_steps: int = 20,
    strict_dependencies: bool = False,
    approve_plan: bool = False,
    response_format: type[BaseModel] | None = None,
    working_memory: WorkingMemory | None = None,
    long_term_memory: LongTermMemory | None = None,
    max_tool_calls: int = 10,
    max_concurrent_tools: int | None = None,
    max_tokens: int = 4096,
    stream: bool = True,
    approval_mode: str = "default",
    on_complete: list[Callable] | None = None,
    on_pause: list[Callable] | None = None,
    reasoning_effort: str | None = None,
) -> task

Parameters most relevant to this page:

The remaining parameters (tools, skills, agents, planning, working_memory, long_term_memory, response_format, max_concurrent_tools, approval_mode, on_complete, on_pause) are covered in the Capabilities and Building blocks sections.

How agent tasks interact with the event log

Each await assistant(...) call is recorded as a Flux task event. If the workflow crashes mid-run, replay resumes from the last recorded checkpoint: the agent’s conversation state is preserved and the LLM is not called again for completed turns. See Agents in workflows for the full durability story, including what happens when tools crash mid-loop.

Next steps

• Agents in workflows — replay semantics and what gets checkpointed.
• Choosing a provider — API key setup, capability matrix, and model-string reference.
• Tools and structured output — giving the agent Flux tasks it can call as tools.
• Agent harness — the standalone YAML-driven harness for long-running interactive agents.