Dynamiq
Agents

ReAct Agent

Configure the SDK's unified Agent class — LLM, role, tools, loops, inference modes, structured output, sandbox, file store, and streaming.

dynamiq.nodes.agents.Agent is the SDK's unified agent: a ReAct-style loop that alternates between reasoning, tool calls, and observations until it produces a final answer. It is the same agent that powers the Agent node on the platform, configured entirely in Python.

Minimal example

An agent needs only an LLM. Tools, memory, and everything else are optional:

from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI

llm = OpenAI(
    connection=OpenAIConnection(),  # reads OPENAI_API_KEY from the environment
    model="gpt-4o",
    temperature=0.3,
)

agent = Agent(
    name="assistant",
    llm=llm,
    role="Helpful assistant that answers questions concisely.",
    max_loops=10,
)

result = agent.run(input_data={"input": "What is the capital of France?"})
print(result.output.get("content"))

Add tools by passing any tool node (or another Agent) in the tools list:

from dynamiq.connections import E2B as E2BConnection, OpenAI as OpenAIConnection
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI
from dynamiq.nodes.tools.e2b_sandbox import E2BInterpreterTool

e2b_tool = E2BInterpreterTool(connection=E2BConnection())  # reads E2B_API_KEY

llm = OpenAI(connection=OpenAIConnection(), model="gpt-4o", temperature=0.3)

agent = Agent(
    name="react-agent",
    llm=llm,
    tools=[e2b_tool],
    role="Senior Data Scientist",
    max_loops=10,
)

result = agent.run(
    input_data={"input": "Add the first 10 numbers and tell if the result is prime."}
)
print(result.output.get("content"))

In production you usually wrap the agent in a Workflow so you get tracing, callbacks, and consistent output handling — see Workflows, Flows & Nodes. Inside a workflow the agent's answer lives at result.output[agent.id]["output"]["content"].

Constructor parameters

llmBaseLLMrequired
The LLM the agent reasons with. Any node from dynamiq.nodes.llms.
rolestr
The agent persona and instructions injected into the system prompt.
toolslist[Node]
Tool nodes the agent may call. Other Agent instances are accepted and run as sub-agents.
max_loopsint
Maximum reasoning loops (default 15, minimum 2).
inference_modeInferenceMode
How the agent formats tool calls: DEFAULT, XML, FUNCTION_CALLING, or STRUCTURED_OUTPUT.
behaviour_on_max_loopsBehavior
Behavior.RAISE (default) raises when max_loops is exceeded; Behavior.RETURN returns the best answer so far.
memoryMemory
Conversation memory — see the Memory page.
response_formatdict | type[BaseModel]
JSON schema (or a Pydantic model class, converted automatically) for structured final output; the final answer is parsed into a dict.
summarization_configSummarizationConfig
Automatic history summarization when the prompt approaches the context window.
file_storeFileStoreConfig
Gives the agent file read/write tools backed by a file storage backend.
sandboxSandboxConfig
Gives the agent an isolated execution environment (E2B or Daytona) with shell and file tools.
parallel_tool_calls_enabledbool
Lets the agent execute multiple tool calls in one loop iteration.
delegation_allowedbool
Lets the agent delegate producing the final answer to a sub-agent.
tool_output_max_lengthint
Token budget for a single tool observation; longer outputs are truncated while tool_output_truncate_enabled is true (default).
streamingStreamingConfig
Streaming configuration — see Streaming below.
verbosebool
Print reasoning steps to the log (default false).

Inference modes

inference_mode controls the protocol between the agent and the LLM:

ModeHow tool calls are expressed
InferenceMode.DEFAULTClassic ReAct text protocol (Thought: / Action: / Action Input:)
InferenceMode.XMLXML tags — robust across providers, a good default for complex tool use
InferenceMode.FUNCTION_CALLINGNative provider function calling
InferenceMode.STRUCTURED_OUTPUTProvider structured-output (JSON schema) responses
from dynamiq.nodes.types import Behavior, InferenceMode

agent = Agent(
    name="researcher",
    llm=llm,
    tools=[e2b_tool],
    inference_mode=InferenceMode.XML,
    behaviour_on_max_loops=Behavior.RETURN,
    max_loops=8,
)

FUNCTION_CALLING and STRUCTURED_OUTPUT require model support; the agent checks provider capabilities through litellm at run time. XML and DEFAULT work with any chat model.

Run-time inputs

agent.run(input_data={...}) (or workflow.run) accepts:

inputstrrequired
The task or question for the agent.
imageslist[str | bytes | BytesIO]
Images for vision-capable LLMs.
fileslist[bytes | BytesIO]
Files made available to the agent and its tools.
user_idstr
Scopes memory reads/writes to a user.
session_idstr
Scopes memory reads/writes to a conversation session.
metadatadict
Arbitrary key-value metadata for the run.
tool_paramsToolParams
Per-run parameter injection into tools (see below).

Injecting tool parameters at run time

tool_params lets the caller pass values (credentials, user context, runtime flags) into tools without the LLM seeing them. Parameters merge with rising precedence: globalby_nameby_id:

result = agent.run(
    input_data={
        "input": "Search for the latest earnings report.",
        "tool_params": {
            "global": {"user_region": "EU"},
            "by_name": {"search-tool": {"num_results": 10}},
            "by_id": {"tool-node-id-123": {"api_key": "value-with-highest-priority"}},
        },
    }
)

Structured output

Set response_format to a JSON schema dict or a Pydantic model class; the agent's final answer is parsed from JSON into a dict:

from pydantic import BaseModel

class Verdict(BaseModel):
    summary: str
    confidence: float

agent = Agent(name="judge", llm=llm, response_format=Verdict)

result = agent.run(input_data={"input": "Is this invoice complete? Items: ..."})
print(result.output["content"])  # {'summary': '...', 'confidence': 0.92}

History summarization

For long-running loops, enable summarization_config so the agent compresses old history instead of overflowing the context window:

from dynamiq.nodes.agents.utils import SummarizationConfig

agent = Agent(
    name="long-task-agent",
    llm=llm,
    tools=[e2b_tool],
    max_loops=40,
    summarization_config=SummarizationConfig(
        enabled=True,
        context_usage_ratio=0.8,      # summarize when 80% of the context window is used
        max_preserved_tokens=10000,   # keep this many tokens of recent messages verbatim
    ),
)
enabledbool
Defaults to false.
max_token_context_lengthint
Absolute token threshold that triggers summarization (used instead of the ratio when set).
context_usage_ratiofloat
Fraction of the model context window that triggers summarization (default 0.8).
max_preserved_tokensint
Token budget for recent messages kept verbatim (default 10000).
token_budget_ratiofloat
Fraction of the context window used for input during the summarization call (default 0.75).

File store and sandbox

file_store gives the agent file tools backed by a storage backend; sandbox gives it a full isolated environment with shell access:

from dynamiq.storages.file.base import FileStoreConfig
from dynamiq.storages.file.in_memory import InMemoryFileStore

agent = Agent(
    name="file-agent",
    llm=llm,
    file_store=FileStoreConfig(
        enabled=True,
        backend=InMemoryFileStore(),
        agent_file_write_enabled=True,  # allow writes, not just reads
        todo_enabled=False,             # optionally add todo-management tools
    ),
)
from dynamiq.connections import E2B as E2BConnection
from dynamiq.sandboxes import SandboxConfig
from dynamiq.sandboxes.e2b import E2BSandbox

sandbox = E2BSandbox(connection=E2BConnection(), timeout=3600, base_path="/home/user")

agent = Agent(
    name="sandbox-agent",
    llm=llm,
    sandbox=SandboxConfig(enabled=True, backend=sandbox),
    max_loops=10,
)

Available sandbox backends: E2BSandbox, DaytonaSandbox, and E2BDesktopSandbox (all in dynamiq.sandboxes). See Sandboxes for backend setup, and the platform equivalents in Sandbox and File Store.

Streaming

Attach a StreamingConfig and a streaming callback handler. StreamingMode.FINAL (the default) streams only the final answer; StreamingMode.ALL also streams reasoning and tool events:

from dynamiq import Workflow
from dynamiq.callbacks.streaming import StreamingIteratorCallbackHandler
from dynamiq.flows import Flow
from dynamiq.runnables import RunnableConfig
from dynamiq.types.streaming import StreamingConfig, StreamingMode

agent = Agent(
    name="research-assistant",
    llm=llm,
    role="Research assistant that explains its reasoning step by step.",
    streaming=StreamingConfig(enabled=True, mode=StreamingMode.ALL),
    max_loops=5,
)

handler = StreamingIteratorCallbackHandler()
wf = Workflow(flow=Flow(nodes=[agent]))
wf.run(
    input_data={"input": "Summarize the pros and cons of vector databases."},
    config=RunnableConfig(callbacks=[handler]),
)

for chunk in handler:
    print(chunk)

See Streaming & Callbacks for the event format and async patterns.

Output

The agent returns a dict with the final answer under content. When the agent requested output files and a file store or sandbox is configured, the collected files are returned under files:

result = agent.run(input_data={"input": "Write a CSV with the first 5 primes and return it."})
answer = result.output["content"]
files = result.output.get("files", [])

Next steps

On this page