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
llmBaseLLMrequiredrolestrtoolslist[Node]max_loopsintinference_modeInferenceModebehaviour_on_max_loopsBehaviormemoryMemoryresponse_formatdict | type[BaseModel]summarization_configSummarizationConfigfile_storeFileStoreConfigsandboxSandboxConfigparallel_tool_calls_enabledbooldelegation_allowedbooltool_output_max_lengthintstreamingStreamingConfigverboseboolInference modes
inference_mode controls the protocol between the agent and the LLM:
| Mode | How tool calls are expressed |
|---|---|
InferenceMode.DEFAULT | Classic ReAct text protocol (Thought: / Action: / Action Input:) |
InferenceMode.XML | XML tags — robust across providers, a good default for complex tool use |
InferenceMode.FUNCTION_CALLING | Native provider function calling |
InferenceMode.STRUCTURED_OUTPUT | Provider 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:
inputstrrequiredimageslist[str | bytes | BytesIO]fileslist[bytes | BytesIO]user_idstrsession_idstrmetadatadicttool_paramsToolParamsInjecting 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: global → by_name → by_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
),
)enabledboolmax_token_context_lengthintcontext_usage_ratiofloatmax_preserved_tokensinttoken_budget_ratiofloatFile 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", [])