Agent

The base agent and its execution context.

agent

The agent programming model for autonomous scientific research.

A Science ADK agent is a Google Agent Development Kit (ADK) agent. Inheriting from google.adk.agents.BaseAgent means experiments natively integrate with ADK sessions, runners, sequential/parallel agent composition, and event streams, while remaining directly executable by deterministic scientific workflows.

Example

Write a single-class scientific agent:

from science_adk import Agent, Port, Ports

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

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

    async def execute(self) -> dict[str, Any]:
        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"]}

Port dataclass

Port(name, type='any', description='', required=True)

A single typed input or output port of an agent.

Attributes:

Name	Type	Description
name	str	The unique name of the port on this agent.
type	str	Data type of the port (must be one of PORT_TYPES).
description	str	Explanation of what this port carries.
required	bool	Whether a value is strictly required on this port.

from_dict classmethod

from_dict(data)

Constructs a Port from a dictionary or string shorthand.

Parameters:

Name	Type	Description	Default
data	dict[str, Any] | str	Dictionary containing port specification or string port name.	required

Returns:

Type	Description
Port	A new Port instance.

Source code in python/src/science_adk/models.py

@classmethod
def from_dict(cls, data: dict[str, Any] | str) -> Port:
    """Constructs a Port from a dictionary or string shorthand.

    Args:
      data: Dictionary containing port specification or string port name.

    Returns:
      A new Port instance.
    """
    if isinstance(data, str):
        return cls(name=data)
    return cls(
        name=data["name"],
        type=data.get("type", "any"),
        description=data.get("description", ""),
        required=bool(data.get("required", True)),
    )

to_dict

to_dict()

Serializes the Port to a dictionary.

Source code in python/src/science_adk/models.py

def to_dict(self) -> dict[str, Any]:
    """Serializes the Port to a dictionary."""
    return asdict(self)

Ports dataclass

Ports(inputs=list(), outputs=list())

The complete input and output port surface of an agent.

Attributes:

Name	Type	Description
inputs	list[Port]	List of declared input ports.
outputs	list[Port]	List of declared output ports.

required_inputs property

required_inputs

Returns all input ports marked as required.

input

input(name)

Retrieves an input port by name, or None if not found.

Source code in python/src/science_adk/models.py

def input(self, name: str) -> Port | None:
    """Retrieves an input port by name, or None if not found."""
    return next((p for p in self.inputs if p.name == name), None)

output

output(name)

Retrieves an output port by name, or None if not found.

Source code in python/src/science_adk/models.py

def output(self, name: str) -> Port | None:
    """Retrieves an output port by name, or None if not found."""
    return next((p for p in self.outputs if p.name == name), None)

from_dict classmethod

from_dict(data)

Constructs Ports from dictionary data.

Parameters:

Name	Type	Description	Default
data	dict[str, Any] | None	Dictionary with 'inputs' and 'outputs' lists.	required

Returns:

Type	Description
Ports	A new Ports instance.

Source code in python/src/science_adk/models.py

@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> Ports:
    """Constructs Ports from dictionary data.

    Args:
      data: Dictionary with 'inputs' and 'outputs' lists.

    Returns:
      A new Ports instance.
    """
    data = data or {}
    return cls(
        inputs=[Port.from_dict(p) for p in data.get("inputs", [])],
        outputs=[Port.from_dict(p) for p in data.get("outputs", [])],
    )

to_dict

to_dict()

Serializes the Ports instance to a dictionary.

Source code in python/src/science_adk/models.py

def to_dict(self) -> dict[str, Any]:
    """Serializes the Ports instance to a dictionary."""
    return {
        "inputs": [p.to_dict() for p in self.inputs],
        "outputs": [p.to_dict() for p in self.outputs],
    }

AgentError

Bases: RuntimeError

Raised when an agent misuses the runtime (bad port, missing input).

Agent

Bases: BaseAgent

Base class for scientific agents built on Google ADK.

Subclasses declare :attr:ports and implement :meth:execute. Everything else exists to make that method short, reproducible, and verifiable.

Attributes:

Name	Type	Description
ports	Ports	Typed input and output port declarations.
params	dict[str, Any]	Static parameters from the experiment workflow definition.
purpose	str	Description of what this agent does and why.
kind	str	Primitive role played by this agent (e.g., algorithm, data).

provenance property

provenance

Returns the machine-recorded provenance ledger for this agent.

logs property

logs

Returns recorded execution log entries.

execute async

execute()

Executes the scientific computation.

Returns:

Type	Description
dict[str, Any]	A dictionary mapping declared output port names to their computed values.

Raises:

Type	Description
NotImplementedError	If the subclass does not implement this method.

Source code in python/src/science_adk/agent.py

async def execute(self) -> dict[str, Any]:
    """Executes the scientific computation.

    Returns:
      A dictionary mapping declared output port names to their computed values.

    Raises:
      NotImplementedError: If the subclass does not implement this method.
    """
    raise NotImplementedError(
        f"{type(self).__name__} must implement `async def execute(self)`."
    )

bind

bind(**inputs)

Provides input values directly.

Parameters:

Name	Type	Description	Default
**inputs	Any	Keyword arguments representing input port values.	{}

Returns:

Type	Description
Agent	Self for chaining.

Source code in python/src/science_adk/agent.py

def bind(self, **inputs: Any) -> Agent:
    """Provides input values directly.

    Args:
      **inputs: Keyword arguments representing input port values.

    Returns:
      Self for chaining.
    """
    self._inputs.update(inputs)
    return self

attach

attach(services)

Attaches execution services (tools, data storage, logging).

Parameters:

Name	Type	Description	Default
services	Any	The execution services provider.	required

Returns:

Type	Description
Agent	Self for chaining.

Source code in python/src/science_adk/agent.py

def attach(self, services: Any) -> Agent:
    """Attaches execution services (tools, data storage, logging).

    Args:
      services: The execution services provider.

    Returns:
      Self for chaining.
    """
    self._services = services
    return self

input async

input(port, default=None)

Reads a declared input port.

Parameters:

Name	Type	Description	Default
port	str	The name of the input port.	required
default	Any	Default value if the port carries no value and is optional.	None

Returns:

Type	Description
Any	The hydrated value associated with the port.

Raises:

Type	Description
AgentError	If the port is not declared or a required port has no value.

Source code in python/src/science_adk/agent.py

async def input(self, port: str, default: Any = None) -> Any:
    """Reads a declared input port.

    Args:
      port: The name of the input port.
      default: Default value if the port carries no value and is optional.

    Returns:
      The hydrated value associated with the port.

    Raises:
      AgentError: If the port is not declared or a required port has no value.
    """
    declared = self.ports.input(port)
    if declared is None:
        names = ", ".join(p.name for p in self.ports.inputs) or "none declared"
        raise AgentError(
            f"{type(self).__name__} read undeclared input port {port!r}. "
            f"Declared inputs: {names}"
        )

    value = self._inputs.get(port)
    if value is not None and self._services is not None:
        value = self._services.hydrate(value)

    found = value is not None
    self._provenance["inputs_read"].append({"port": port, "found": found})

    if not found:
        if declared.required and default is None:
            raise AgentError(
                f"Required input {port!r} of {self.name!r} has no value. "
                "Ensure an edge in workflow.json feeds this port."
            )
        return default
    return value

param

param(name, default=None)

Reads a static parameter from node configuration.

Parameters:

Name	Type	Description	Default
name	str	Name of the parameter.	required
default	Any	Default fallback value if not specified.	None

Returns:

Type	Description
Any	The parameter value.

Source code in python/src/science_adk/agent.py

def param(self, name: str, default: Any = None) -> Any:
    """Reads a static parameter from node configuration.

    Args:
      name: Name of the parameter.
      default: Default fallback value if not specified.

    Returns:
      The parameter value.
    """
    return self.params.get(name, default)

call_tool async

call_tool(name, **arguments)

Calls a tool by name and records provenance.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool.	required
**arguments	Any	Arguments to pass to the tool.	{}

Returns:

Type	Description
Any	The tool execution result.

Raises:

Type	Description
AgentError	If execution services are not attached.

Source code in python/src/science_adk/agent.py

async def call_tool(self, name: str, **arguments: Any) -> Any:
    """Calls a tool by name and records provenance.

    Args:
      name: Name of the tool.
      **arguments: Arguments to pass to the tool.

    Returns:
      The tool execution result.

    Raises:
      AgentError: If execution services are not attached.
    """
    if self._services is None:
        raise AgentError(
            f"{self.name}: no tools are attached. This agent is not being "
            "run within an active runtime context."
        )
    try:
        result = await self._services.call_tool(name, arguments)
    except Exception:
        self._provenance["tool_calls"].append({"tool": name, "ok": False})
        raise
    self._provenance["tool_calls"].append({"tool": name, "ok": True})
    return result

fetch async

fetch(url, timeout=60, retries=3)

Fetches a URL with bounded exponential backoff and records provenance.

Parameters:

Name	Type	Description	Default
url	str	The URL to fetch.	required
timeout	int	Timeout in seconds.	60
retries	int	Maximum number of retry attempts.	3

Returns:

Type	Description
str	The decoded response text.

Raises:

Type	Description
AgentError	If the fetch fails after all retry attempts.

Source code in python/src/science_adk/agent.py

async def fetch(self, url: str, timeout: int = 60, retries: int = 3) -> str:
    """Fetches a URL with bounded exponential backoff and records provenance.

    Args:
      url: The URL to fetch.
      timeout: Timeout in seconds.
      retries: Maximum number of retry attempts.

    Returns:
      The decoded response text.

    Raises:
      AgentError: If the fetch fails after all retry attempts.
    """
    last: Exception | None = None
    for attempt in range(retries):
        try:
            request = urllib.request.Request(
                url, headers={"User-Agent": "science-adk/0.1"}
            )
            body = await asyncio.to_thread(_read_url, request, timeout)
            self._provenance["urls_fetched"].append({"url": url, "ok": True})
            return body
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            last = exc
            if attempt < retries - 1:
                await asyncio.sleep(2**attempt)
    self._provenance["urls_fetched"].append({"url": url, "ok": False})
    raise AgentError(f"Failed to fetch {url} after {retries} attempts: {last}")

artifact

artifact(filename)

Reserves a path in this run's data directory and logs file creation.

Parameters:

Name	Type	Description	Default
filename	str	Name of the artifact file.	required

Returns:

Type	Description
Path	Path to the reserved file location.

Raises:

Type	Description
AgentError	If execution services are not attached.

Source code in python/src/science_adk/agent.py

def artifact(self, filename: str) -> Path:
    """Reserves a path in this run's data directory and logs file creation.

    Args:
      filename: Name of the artifact file.

    Returns:
      Path to the reserved file location.

    Raises:
      AgentError: If execution services are not attached.
    """
    if self._services is None:
        raise AgentError(
            f"{self.name}: no run directory attached. This agent is not "
            "being run within an active runtime context."
        )
    directory = self._services.data_dir()
    directory.mkdir(parents=True, exist_ok=True)
    self._provenance["files_written"].append({"file": filename})
    return directory / filename

log async

log(message)

Records an informational log message in the provenance trace.

Parameters:

Name	Type	Description	Default
message	str	Message text to record.	required

Source code in python/src/science_adk/agent.py

async def log(self, message: str) -> None:
    """Records an informational log message in the provenance trace.

    Args:
      message: Message text to record.
    """
    text = str(message)
    self._logs.append(text)
    if self._services is not None:
        await self._services.log(self.name, text)

validate_outputs

validate_outputs(outputs)

Validates execution output against declared output ports.

Parameters:

Name	Type	Description	Default
outputs	Any	The output dictionary returned by execute.	required

Returns:

Type	Description
dict[str, Any]	The validated outputs dictionary.

Raises:

Type	Description
AgentError	If output types or keys mismatch declared output ports.

Source code in python/src/science_adk/agent.py

def validate_outputs(self, outputs: Any) -> dict[str, Any]:
    """Validates execution output against declared output ports.

    Args:
      outputs: The output dictionary returned by `execute`.

    Returns:
      The validated outputs dictionary.

    Raises:
      AgentError: If output types or keys mismatch declared output ports.
    """
    name = type(self).__name__
    if not isinstance(outputs, dict):
        raise AgentError(
            f"{name}.execute must return a dict of output ports, "
            f"got {type(outputs).__name__}."
        )
    declared = {p.name for p in self.ports.outputs}
    missing = sorted(declared - set(outputs))
    if missing:
        raise AgentError(
            f"{name}.execute did not return declared output port(s):"
            f" {', '.join(missing)}."
        )
    extra = sorted(set(outputs) - declared)
    if extra:
        raise AgentError(
            f"{name}.execute returned undeclared output(s): {', '.join(extra)}. "
            "Add them to `ports` or remove them."
        )
    for port in self.ports.outputs:
        if outputs[port.name] is None and port.required:
            raise AgentError(
                f"{name}.execute returned None for required output {port.name!r}."
            )
    return outputs

default_ports classmethod

default_ports()

Returns default ports declared on the class without instantiation.

Source code in python/src/science_adk/agent.py

@classmethod
def default_ports(cls) -> Ports:
    """Returns default ports declared on the class without instantiation."""
    return _field_default(cls, "ports") or Ports()

describe classmethod

describe()

Returns a machine-readable summary of the agent class.

Source code in python/src/science_adk/agent.py

@classmethod
def describe(cls) -> dict[str, Any]:
    """Returns a machine-readable summary of the agent class."""
    purpose = _field_default(cls, "purpose") or ""
    return {
        "class": cls.__name__,
        "kind": cls.kind,
        "purpose": purpose or (cls.__doc__ or "").strip().split("\n")[0],
        "ports": cls.default_ports().to_dict(),
    }

load_agent_class

load_agent_class(path)

Imports a single-class agent file and returns its Agent subclass.

Parameters:

Name	Type	Description	Default
path	Path	Path to the agent Python file.	required

Returns:

Type	Description
type[Agent]	The single Agent subclass defined in the file.

Raises:

Type	Description
AgentError	If the file does not exist, fails to import, or defines zero or multiple Agent subclasses.

Source code in python/src/science_adk/agent.py

def load_agent_class(path: Path) -> type[Agent]:
    """Imports a single-class agent file and returns its Agent subclass.

    Args:
      path: Path to the agent Python file.

    Returns:
      The single Agent subclass defined in the file.

    Raises:
      AgentError: If the file does not exist, fails to import, or defines zero or
        multiple Agent subclasses.
    """
    if not path.exists():
        raise AgentError(f"Agent file not found: {path}")

    spec = importlib.util.spec_from_file_location(
        f"_science_adk_agent_{path.stem}", path
    )
    if spec is None or spec.loader is None:
        raise AgentError(f"Cannot load agent file: {path}")
    module = importlib.util.module_from_spec(spec)
    try:
        spec.loader.exec_module(module)
    except Exception as exc:
        raise AgentError(
            f"Error importing {path.name}: {type(exc).__name__}: {exc}"
        ) from exc

    candidates = [
        obj
        for obj in vars(module).values()
        if isinstance(obj, type)
        and issubclass(obj, Agent)
        and obj is not Agent
        and obj.__module__ == module.__name__
    ]
    leaves = [
        c
        for c in candidates
        if not any(other is not c and issubclass(other, c) for other in candidates)
    ]

    if not leaves:
        raise AgentError(
            f"{path.name} defines no Agent subclass. "
            "An agent file must contain exactly one class inheriting from Agent."
        )
    if len(leaves) > 1:
        names = ", ".join(sorted(c.__name__ for c in leaves))
        raise AgentError(
            f"{path.name} defines {len(leaves)} Agent subclasses ({names}). "
            "An agent file must contain exactly one."
        )
    return leaves[0]