Skip to content

Agents and typed ports

An agent is one node of an experiment: a Python class that declares what it consumes, what it produces, and how to get from one to the other.

from science_adk import Agent, Port, Ports


class FitOscillator(Agent):
    """Recover oscillator parameters from a simulated trajectory."""

    ports: Ports = Ports(
        inputs=[Port("trajectory", "dataframe")],
        outputs=[Port("period", "float"), Port("residual", "float")],
    )

    async def execute(self) -> dict:
        df = await self.input("trajectory")
        await self.log(f"fitting {len(df)} samples")
        result = await self.call_tool("fit_sinusoid", t=list(df.t), y=list(df.theta))
        return {"period": result["period"], "residual": result["rmse"]}

That is the whole programming model. Everything else in the class exists to keep execute short, reproducible and verifiable.

Ports

A Port has four fields:

Field Default Meaning
name — Unique on this agent; also the key used in edges and returns.
type "any" One of the port types below.
description "" What the value carries. Worth writing.
required True If False, input() may return the default instead.

The full type vocabulary:

int · float · bool · str · json · list · dataframe · array · file · figure · model · any

An unknown type raises ValueError at construction, so a typo cannot survive to run time.

Compatibility

An edge is legal when the source port type can feed the target port type:

types_compatible(source, target)
Case Compatible
Same type (json → json) yes
Either side is any yes
int → float yes
float → int no
Everything else (json → dataframe) no

The rule is deliberately narrow. Silent coercion between a dataframe and JSON is exactly the kind of convenience that turns into a data bug three experiments later. science-adk validate reports incompatible edges as errors before anything executes.

The execution contract

execute must return a dict keyed by the declared output port names. validate_outputs rejects a non-dict return and reports missing ports by name, so a node cannot half-produce its declared surface.

run is reserved

Defining def run on an Agent subclass raises TypeError at class creation. run belongs to the Google ADK runner protocol; the scientific entry point is execute.

What an agent can do

Call Purpose
await self.input(port, default=None) Read a declared input port. Undeclared port → AgentError.
self.param(name, default=None) Read a static parameter from the node's params in workflow.json.
await self.call_tool(name, **arguments) Invoke a registered tool. Recorded in provenance.
await self.fetch(url, timeout=60, retries=3) HTTP GET with bounded exponential backoff. Recorded in provenance.
self.artifact(filename) Reserve a path in this run's data/ directory. Recorded in provenance.
await self.log(message) Append a line to the node's trace log.
self.provenance, self.logs Read back what has been recorded.

Each of call_tool, fetch and artifact writes to the provenance ledger as a side effect the agent's own code does not control — see provenance.

Inputs arrive by port name

The runner binds each incoming edge to the target port it names. Inside execute, await self.input("trajectory") returns that value, hydrated from disk first if it travelled as a dataset reference.

Reading a port that was never declared is an error, not a None:

AgentError: FitOscillator read undeclared input port 'trajectry'.
Declared inputs: trajectory

A required port with no value is also an error, and it names the fix:

AgentError: Required input 'trajectory' of 'fit' has no value.
Ensure an edge in workflow.json feeds this port.

It really is a Google ADK agent

Agent subclasses google.adk.agents.BaseAgent and implements _run_async_impl, emitting a standard Event with the validated outputs attached as custom_metadata. Anything in the Google ADK ecosystem that orchestrates BaseAgent instances can run a Science ADK agent unchanged:

from google.adk.runners import InMemoryRunner
from science_adk import Agent, Port, Ports


class ScienceAgent(Agent):
    ports: Ports = Ports(inputs=[], outputs=[Port("value", "int")])

    async def execute(self):
        return {"value": 42}


runner = InMemoryRunner(agent=ScienceAgent(name="science_agent"))

See built on Google ADK for the measurements behind that decision.

Next