Write an agent node¶

Every node in workflow.json needs exactly one Python file containing exactly one Agent subclass. This guide covers the whole surface and the errors you will hit if you get it wrong.

The file¶

By default the runner looks for agents/<node id>.py inside the experiment directory; override it with the node's module field.

research/001-.../agents/analyze.py

"""Fit the period–semi-major-axis power law to recover Kepler's exponent."""

from typing import ClassVar

from science_adk import AlgorithmAgent, Port, Ports


class FitKeplerLaw(AlgorithmAgent):
    """Recover the power-law exponent from a blind log-log fit.

    The fit is ordinary least squares in log-log space and makes no
    assumption about the exponent. If the data followed T ∝ a², the tool
    would report that. The match to 3/2 is a result, not a tautology.
    """

    assumptions: ClassVar[list[str]] = [
        "The log-log linear model is appropriate for this relationship.",
        "Host star mass variation adds scatter but does not bias the slope.",
    ]

    ports: Ports = Ports(
        inputs=[Port("planets", "json")],
        outputs=[Port("fit_result", "json", "Exponent, r², diagnostics")],
    )

    async def execute(self):
        planets = await self.input("planets")

        periods = [p["period_days"] for p in planets]
        axes = [p["semi_major_axis_au"] for p in planets]

        fit = await self.call_tool("power_law_fit", x=axes, y=periods)

        await self.log(
            f"fit {fit['n']} planets: T ∝ a^{fit['exponent']:.4f}, "
            f"r² = {fit['r_squared']:.6f}"
        )

        return {"fit_result": {
            "exponent": fit["exponent"],
            "r_squared": fit["r_squared"],
            "n_planets": fit["n"],
        }}

Four rules, all enforced statically:

One agent class per file.
ports is annotated: ports: Ports = Ports(...).
execute exists and is async.
The class's port names match workflow.json exactly.

Choosing a base class¶

Pick the primitive matching the node's kind:

from science_adk import (
    AlgorithmAgent,      # kind: algorithm — the method under test
    ConfigAgent,         # kind: config
    DataAgent,           # kind: data
    EvaluationAgent,     # kind: evaluation
    TrainingAgent,       # kind: training
    VisualizationAgent,  # kind: visualization
    CompositeAgent,      # kind: composite
    Agent,               # anything else
)

Reading inputs¶

planets = await self.input("planets")              # required: raises if absent
notes   = await self.input("notes", default=[])    # optional port
threads = self.param("threads", 4)                 # static, from node params

input() hydrates dataset references transparently, so a dataframe port yields a dataframe, not a pointer.

Reading a port you did not declare is an error rather than a None:

AgentError: FitKeplerLaw read undeclared input port 'planet'.
Declared inputs: planets

Calling tools¶

fit = await self.call_tool("power_law_fit", x=axes, y=periods)

Then declare it on the node so the promise is checked:

{ "id": "analyze", "kind": "algorithm", "tools": ["power_law_fit"] }

If two providers export the same name, qualify it: await self.call_tool("local.power_law_fit", ...).

Fetching data directly¶

body = await self.fetch(url, timeout=60, retries=3)

Bounded exponential backoff, and every attempt recorded in provenance. Prefer a tool when the request has any real logic in it — a tool is reusable, testable and independently documented.

Logging and artifacts¶

await self.log(f"kept {len(clean)} of {len(raw)} rows after QC")

path = self.artifact("residuals.csv")     # reserved inside the run's data/
frame.to_csv(path, index=False)

Logs land in the node's trace entry, so they are part of the permanent record rather than terminal scrollback. Log the numbers a reviewer will want: sample counts, ranges, what was discarded and why.

Returning outputs¶

Return a dict keyed by the declared output port names — all of them:

return {"fit_result": {...}}

AgentError: FitKeplerLaw.execute must return a dict of output ports, got list.

For an evaluation node, use report() instead of building the dict by hand; it rejects NaN and non-numeric values at source:

return self.report(
    metric="r_squared",
    value=fit["r_squared"],
    detail={"recovered_exponent": exponent, "n_planets": fit["n_planets"]},
)

What not to do¶

Never swallow an exception

try:
    fit = await self.call_tool("power_law_fit", x=axes, y=periods)
except Exception:
    fit = {"exponent": 1.5, "r_squared": 1.0}   # (1)!

validate reports this as an error: "exception is caught and discarded — let it propagate. A swallowed error becomes a fabricated result." A failed run is information; a fake success is not.

Other patterns that draw a warning:

Pattern	Why it is flagged
bare `except:`	Catches your own bugs along with the error you meant to handle.
`eval()` / `exec()`	Executes arbitrary strings. Compute the value instead.
`warnings.filterwarnings(...)`	Numerical warnings usually mean a real problem.
Unannotated class attributes	Agents are pydantic models. Use `ClassVar[...]`.
A method named `run`	Reserved by the Google ADK runner protocol. Use `execute`.

Verify¶

science-adk validate
science-adk run --only analyze

--only runs just that node and its dependents, which makes the write–check loop fast while you are still iterating.

Testing an agent directly¶

Because it is a real Google ADK agent, it runs under a standard runner:

from google.adk.runners import InMemoryRunner
from agents.analyze import FitKeplerLaw

agent = FitKeplerLaw(name="analyze").bind(planets=[...])
runner = InMemoryRunner(agent=agent)