Skip to content

Agents

consortium.framework.agents

The agents framework provides the building blocks for defining agents and the commands they expose.

An agent is described by a BaseAgentType, which groups the BaseAgentCapability commands the agent supports. Deployable agent artifacts are produced by a BaseAgentGenerator, configured and created through a BaseAgentTemplate. Capabilities communicate with a live agent using the task message models (TaskLaunchMessageModel, TaskInputMessageModel, TaskOutputMessageModel) and report results as a Success or Failure outcome.

PayloadTooLargeError(max_size)

Bases: AgentCapabilitiesFrameworkError

Raised when a payload being received exceeds a maximum size its transport set.

The cap is opt in: Payload.from_async_iterable is uncapped by default and enforces a bound only when a caller passes max_size, in which case it is enforced as the source is consumed so an unbounded upload is rejected mid-stream rather than after being fully buffered. This is the single place transports learn a payload was refused for size, replacing the per-transport sentinels and cap loops each one used to carry.

code = 'PAYLOAD_TOO_LARGE_ERROR' class-attribute instance-attribute

Payload(*, data=None, spool=None, size, resident_size=None, filename=None, content_type=None)

A binary attachment on a task message.

Read a payload whole or stream it a chunk at a time, either of them any number of times:

data = await payload.read()
async for chunk in payload: ...

Construct one from bytes already in hand, from a file, from an open file object, or from an async byte source such as an upload:

Payload.from_bytes(data)
await Payload.from_file(path)
await Payload.from_file_like(fileobj)
await Payload.from_async_iterable(source)

Every constructor takes a copy of its source, so a payload never refers back to it and is unaffected by what happens to it afterwards. A large payload is held on disk rather than in memory, and is cleaned up once nothing refers to the payload, so callers never manage its storage.

Attributes:

Name Type Description
size

The size of the payload in bytes, known before it is read.
filename

The originating filename, if known, or None.
content_type

The MIME content type, if known, or None.

size = size instance-attribute

filename = filename instance-attribute

content_type = content_type instance-attribute

from_bytes(data, *, filename=None, content_type=None) classmethod

Build an in-memory payload from a complete byte sequence.

Parameters:

Name Type Description Default
data bytes | bytearray

The payload bytes. A bytearray is copied to immutable bytes.

 required
filename str | None

Optional originating filename.

 None
content_type str | None

Optional MIME content type.

 None

Returns:

Type Description
Payload

An in-memory payload holding the given bytes.

from_async_iterable(source, *, max_size=None, spool_to_disk_above=SPOOL_TO_DISK_ABOVE, filename=None, content_type=None) async classmethod

Consume an async byte source into a payload.

When a size cap is given it is enforced as bytes arrive, so an oversized source is rejected without being read to the end.

Parameters:

Name Type Description Default
source AsyncIterable[bytes]

An async iterable of byte chunks, such as an upload being received.

 required
max_size int | None

Optional hard maximum total size in bytes, raising PayloadTooLargeError when exceeded. Defaults to None, meaning no cap: the framework imposes no payload size of its own, so a transport that wants one passes its own here.

 None
spool_to_disk_above int

Size in bytes above which the payload is held on disk instead of in memory. Defaults to the framework's configured threshold.

 SPOOL_TO_DISK_ABOVE
filename str | None

Optional originating filename.

 None
content_type str | None

Optional MIME content type.

 None

Raises:

Type Description
PayloadTooLargeError

If the source yields more than max_size bytes.

Returns:

Type Description
Payload

The received payload.

from_file(path, *, max_size=None, spool_to_disk_above=SPOOL_TO_DISK_ABOVE, filename=None, content_type=None) async classmethod

Build a payload from a file on disk, copying its contents.

The file is read once, now, into the payload's own storage. The payload is a snapshot of the file as it was at this moment and never refers to it again, so the original may be modified, moved or deleted immediately afterwards, and is not held open in the meantime.

A file at or below spool_to_disk_above is held in memory, so a small file costs a read rather than a copy on disk.

Parameters:

Name Type Description Default
path str | PathLike[str]

The file to read.

 required
max_size int | None

Optional hard maximum total size in bytes, raising PayloadTooLargeError when exceeded. Defaults to None, meaning no cap.

 None
spool_to_disk_above int

Size in bytes above which the payload is held on disk instead of in memory. Defaults to the framework's configured threshold.

 SPOOL_TO_DISK_ABOVE
filename str | None

Originating filename recorded on the payload. Defaults to the name of path.

 None
content_type str | None

Optional MIME content type.

 None

Raises:

Type Description
OSError

If the file cannot be opened or read.
PayloadTooLargeError

If the file holds more than max_size bytes.

Returns:

Type Description
Payload

The payload.

from_file_like(fileobj, *, max_size=None, spool_to_disk_above=SPOOL_TO_DISK_ABOVE, filename=None, content_type=None) async classmethod

Build a payload from an open binary file object, copying its contents.

Reads from the object's current position until it is exhausted and copies what it yields into the payload's own storage. The object is not rewound first, not closed afterwards, and not referred to again once this returns.

Parameters:

Name Type Description Default
fileobj IO[bytes]

A file object opened in binary mode, such as an open file or a BytesIO.

 required
max_size int | None

Optional hard maximum total size in bytes, raising PayloadTooLargeError when exceeded. Defaults to None, meaning no cap.

 None
spool_to_disk_above int

Size in bytes above which the payload is held on disk instead of in memory. Defaults to the framework's configured threshold.

 SPOOL_TO_DISK_ABOVE
filename str | None

Optional originating filename.

 None
content_type str | None

Optional MIME content type.

 None

Raises:

Type Description
OSError

If the object cannot be read.
PayloadTooLargeError

If the object yields more than max_size bytes.

Returns:

Type Description
Payload

The payload.

read() async

Read the whole payload and return it as a single byte sequence.

This is the point at which a payload is fully materialized in memory, including a payload that spilled to disk and has been costing a queue nothing since. The bytes returned sit outside every queue's memory accounting, and a payload arrives as large as its transport allowed, so what is affordable here is the caller's own bound rather than one the framework imposes. Prefer async for chunk in payload unless the whole payload is genuinely needed at once.

Returns:

Type Description
bytes

The complete payload as a contiguous byte sequence.

TaskInputMessageModel

Bases: _TaskMessage

Message sent to an agent to provide additional input to a running task.

Used when a task requires interactive or incremental input after the initial launch message has been sent.

Attributes:

Name Type Description
task_id UUID4

Unique identifier of the task receiving the input.
data dict[str, JsonValue]

Structured input data for the running task. Must be JSON-serializable.
payload Annotated[Payload | None, BeforeValidator(_wrap_payload)]

Optional binary payload accompanying the input, such as a file chunk or continuation data.

data = {} class-attribute instance-attribute

TaskLaunchMessageModel

Bases: _TaskMessage

Message sent to an agent to initiate a new task execution.

Carries the task identity, the command name, and any structured arguments and data the capability needs. An optional binary payload can accompany the message for capabilities that require file or binary input.

Attributes:

Name Type Description
task_id UUID4

Unique identifier for the task being launched.
command str

The name of the capability command the agent should execute.
arguments dict[str, JsonValue]

Command arguments required to execute the capability. Must be JSON-serializable.
data dict[str, JsonValue]

Supplementary data associated with the task. Must be JSON-serializable.
payload Annotated[Payload | None, BeforeValidator(_wrap_payload)]

Optional binary payload accompanying the task, such as a file to be processed by the agent.

command instance-attribute

arguments = {} class-attribute instance-attribute

data = {} class-attribute instance-attribute

TaskOutputMessageModel

Bases: _TaskMessage

Message sent from an agent reporting the result of a completed task.

Carries whether the task succeeded, a human-readable result summary, and any structured output data or binary artifacts produced during execution.

Attributes:

Name Type Description
task_id UUID4

Unique identifier of the task that produced this output.
success bool

True if the task completed successfully, False on failure.
message str

Human-readable summary of the task result or error description.
data dict[str, JsonValue]

Structured output data from the task. Must be JSON-serializable.
payload Annotated[Payload | None, BeforeValidator(_wrap_payload)]

Optional binary artifact produced by the task, such as a captured file or command output blob.

success instance-attribute

message = '' class-attribute instance-attribute

data = {} class-attribute instance-attribute

to_outcome()

Convert the message into a Success or Failure outcome object.

Returns:

Type Description
Success | Failure

A Success instance if success is True, otherwise a Failure instance.

Failure(task_output_message=None, message=None, data=None)

Represents a failed outcome from an agent capability execution.

Carries a message and structured diagnostic data extracted from a TaskOutputMessageModel or supplied directly. Individual fields can be overridden when constructing from a task output message.

Attributes:

Name Type Description
message str

Human-readable description of the failure.
data dict[str, JsonValue]

Structured diagnostic data from the failed execution.

Initialize a Failure outcome from a task output message or explicit values.

When task_output_message is provided, message and data are sourced from it unless overridden by the corresponding keyword arguments.

Parameters:

Name Type Description Default
task_output_message TaskOutputMessageModel | None

The raw output message received from the agent. When provided, message and data default to the values from this model.

 None
message str | None

Human-readable description of the failure. Overrides the message from task_output_message when both are provided.

 None
data dict[str, JsonValue] | None

Structured diagnostic data. Overrides the data from task_output_message when both are provided.

 None

message instance-attribute

data instance-attribute

Success(task_output_message=None, message=None, data=None)

Represents a successful outcome from an agent capability execution.

Carries a message and structured data extracted from a TaskOutputMessageModel or supplied directly. Individual fields can be overridden when constructing from a task output message.

Attributes:

Name Type Description
message str

Human-readable description of the successful result.
data dict[str, JsonValue]

Structured result data from the execution.

Initialize a Success outcome from a task output message or explicit values.

When task_output_message is provided, message and data are sourced from it unless overridden by the corresponding keyword arguments.

Parameters:

Name Type Description Default
task_output_message TaskOutputMessageModel | None

The raw output message received from the agent. When provided, message and data default to the values from this model.

 None
message str | None

Human-readable description of the result. Overrides the message from task_output_message when both are provided.

 None
data dict[str, JsonValue] | None

Structured result data. Overrides the data from task_output_message when both are provided.

 None

message instance-attribute

data instance-attribute

BaseAgentCapability(agent, task)

Bases: _AgentCommunicator

Base class for all agent capabilities that define executable commands.

Subclasses declare capability metadata as class attributes (name, description, options, etc.) and override on_launch and on_execute to control how the command is transmitted to the agent and how the response is processed. The framework validates all class attributes at subclass definition time via init_subclass.

Attributes:

Name Type Description
name str

Unique command identifier used to route incoming task messages. Required and must be non-empty.
description str

Human-readable explanation of what this capability does.
authors set[str]

Identifiers for the capability's authors.
requires_admin bool

Whether elevated privileges are required on the target system to execute this capability.
supported_oses set[SupportedOS]

Platforms this capability supports. Defaults to {SupportedOS.ANY} if not declared.
options set[SingleValueOption | ListValueOption | DictionaryValueOption | ChoiceValueOption | ToggleableChoicesValueOption]

Configuration options accepted by this capability. Declared as a set at the class level; converted to a name-keyed dict at definition time.
mitre_attack_techniques set[str] | None

MITRE ATT&CK technique IDs associated with this capability. Resolved to MitreAttackTechnique objects at definition time.
validating_function Callable[[dict[str, JsonValue]], None] | None

Optional single-argument callable that validates the full resolved option set before execution.
task_launch_message TaskLaunchMessageModel | None

The message sent to the agent on the most recent execute() call; set by execute() after on_launch completes.

Initialize the capability with the agent and task context for this execution.

Parameters:

Name Type Description Default
agent Agent

The agent instance this capability is executing against.

 required
task Task

The task record that tracks the execution lifecycle and event stream.

 required

name instance-attribute

description = '' class-attribute instance-attribute

authors = None class-attribute instance-attribute

requires_admin = False class-attribute instance-attribute

supported_oses = None class-attribute instance-attribute

options = None class-attribute instance-attribute

mitre_attack_techniques = None class-attribute instance-attribute

validating_function = None class-attribute instance-attribute

task_launch_message = None class-attribute instance-attribute

agent_file_manager_service = AgentFileManagerService(agent=agent) instance-attribute

environment = SimpleNamespace() instance-attribute

logger property

System logger for reporting this capability's execution as it runs.

Returns:

Type Description

The system logger shared with the owning task.

event_logger property

Event logger for reporting this capability's execution as it runs.

Shared with the owning task, so events recorded here (success, failure, info, warning, error, artifact, and progress updates) appear in the task's event log and are surfaced to the client. Entries are also mirrored to the task's system logger.

Returns:

Type Description

The event logger shared with the owning task.

send_to_agent(task_message=None, data=None, payload=None, timeout=None) async

Send a message to the agent.

If a prepared task message is provided it is forwarded to the agent as-is. Otherwise a TaskInputMessageModel is built from the current task and the supplied data and payload before being sent.

Parameters:

Name Type Description Default
task_message TaskInputMessageModel | None

A prepared launch or input message to forward to the agent. If provided, data and payload are ignored.

 None
data dict[str, Any] | None

Structured data to include when building a task input message. Only used when task_message is None. Defaults to an empty dictionary.

 None
payload bytes | bytearray | Payload | None

Optional binary payload to attach when building a task input message. Only used when task_message is None.

 None
timeout int | float | None

Maximum number of seconds to wait for the send to complete. If None, waits indefinitely.

 None

recv_from_agent(timeout=None) async

Wait for and return the next message from the agent.

Blocks until the next result message for this task is available on the agent's result messages queue.

Parameters:

Name Type Description Default
timeout int | float | None

Maximum number of seconds to wait for a message. If None, waits indefinitely.

 None

Returns:

Type Description
TaskOutputMessageModel

The next task output message received from the agent.

Raises:

Type Description
TimeoutError

If timeout is set and no message arrives within it.
AgentCommunicationEndOfStreamError

If the inbox reaches end of stream while waiting. A communicator is coordinated with the remote endpoint, so this should never happen during normal operation.

send_and_recv_from_agent(task_message=None, data=None, payload=None, timeout=None) async

Send a message to the agent and wait for its response.

Combines send_to_agent and recv_from_agent into a single round trip. When a timeout is provided, it applies to the combined send-and-receive operation.

Parameters:

Name Type Description Default
task_message TaskInputMessageModel | None

A prepared launch or input message to forward to the agent. If provided, data and payload are ignored.

 None
data dict[str, Any] | None

Structured data to include when building a task input message. Only used when task_message is None.

 None
payload bytes | bytearray | Payload | None

Optional binary payload to attach when building a task input message. Only used when task_message is None.

 None
timeout int | float | None

Maximum number of seconds to wait for the combined send and receive to complete. If None, waits indefinitely.

 None

Returns:

Type Description
TaskOutputMessageModel

The task output message received from the agent in response.

Raises:

Type Description
TimeoutError

If timeout is set and the combined operation does not complete within it.

on_launch(task_launch_message) async

Hook called before the task message is transmitted to the agent.

Override to mutate or enrich the launch message prior to sending. This must return a TaskLaunchMessageModel. To deny the launch (for example when a pre-launch validation check fails) raise AgentCapabilityLaunchError; the task is then reported as ERRORED. Returning anything else is a contract violation rather than a denial and is reported as a fatal error against the capability.

Parameters:

Name Type Description Default
task_launch_message TaskLaunchMessageModel

The task launch message prepared by the caller, containing the command, arguments, data, and any attached payload.

 required

Returns:

Type Description
TaskLaunchMessageModel

The (possibly modified) task message to send.

on_execute() async

Hook called after the task message has been sent to process the agent's response.

Override to implement custom response handling logic. The default implementation waits for a single reply from the agent and wraps it in a Success or Failure.

Returns:

Type Description
Success | Failure | None

A Success wrapping the agent's response on success, a Failure on failure,
Success | Failure | None

or None if no response is expected.

execute(task_launch_message) async

Dispatch the task to the agent and return the execution outcome.

Calls on_launch to allow pre-send mutation, transmits the (possibly modified) message to the agent, then calls on_execute to await and process the response.

Parameters:

Name Type Description Default
task_launch_message TaskLaunchMessageModel

The fully populated task launch message to dispatch, including the command, arguments, data, and any binary payload.

 required

Returns:

Type Description
Success | Failure | None

The outcome from on_execute. Return Success or Failure to opt in to an
Success | Failure | None

explicit terminal event and task transition; return None when the capability
Success | Failure | None

reported everything it needs to through the event logger, in which case the
Success | Failure | None

task is assumed to have completed normally.

Raises:

Type Description
AgentCapabilityLaunchError

If on_launch denies the launch by raising it. The task handler converts this into an ERRORED task.
AgentCapabilityFatalError

If an unexpected exception escapes on_launch, the dispatch of the launch message or on_execute, or if on_launch returns anything other than a TaskLaunchMessageModel. The phase identifies which of those stages failed.

to_json() classmethod

Serialize the capability's class-level metadata to a JSON-compatible dictionary.

Returns:

Type Description
dict[str, Any]

A dictionary containing the capability name, description, authors, options
dict[str, Any]

(with their validation schemas), MITRE ATT&CK techniques, supported OSes,
dict[str, Any]

admin requirement flag, and any validating function documentation.

SupportedOS

Bases: StrEnum

Operating system identifiers for declaring capability platform compatibility.

Use these values in BaseAgentCapability.supported_oses to restrict which platforms a capability can run on. ANY indicates universal compatibility with no restrictions. The DESKTOP and MOBILE class attributes provide pre-built sets of related OS values for convenience.

WINDOWS = 'WINDOWS' class-attribute instance-attribute

LINUX = 'LINUX' class-attribute instance-attribute

MACOS = 'MACOS' class-attribute instance-attribute

ANDROID = 'ANDROID' class-attribute instance-attribute

IOS = 'IOS' class-attribute instance-attribute

ANY = 'ANY' class-attribute instance-attribute

DESKTOP instance-attribute

MOBILE instance-attribute

BaseAgentGenerator(name=None, description='', parameters=None)

Bases: ComponentLifeCycle

Orchestrates a pipeline of build steps to produce a deployable agent payload.

Subclasses declare a list of BaseAgentGeneratorBuildStep classes that are instantiated and executed sequentially at run time. A shared SimpleNamespace environment allows earlier steps to pass state (file paths, signing keys, compiled artifacts, etc.) to later ones.

Attributes:

Name Type Description
agent_generator_build_steps list[BaseAgentGeneratorBuildStep]

Ordered sequence of build step classes. Declared at the class level and converted to instances in init.
name

Human-readable label for this generator run. Either supplied explicitly at creation time or randomly generated. It is display metadata only and is never derived from, or kept in sync with, the generator's parameters.
description

Optional description of the generator run.
parameters

Configuration values forwarded to every build step.
agent_generator_id

Unique identifier for this generator instance.
agent_templates_payload_service

Service used to access agent-template payloads.
datetime_created

UTC timestamp at which this generator instance was created.
environment

Shared mutable namespace available to all build steps.
logger

System logger for this generator.
event_logger

Generator-wide event logger. Every build step is given a child logger that shares this logger's underlying event log, so all build steps report into one consolidated, client-facing event log. Entries are optionally mirrored to the relevant system logger.
root_directory

Directory containing the concrete generator's source file.
services

Server services exposed to the concrete generator.
agent_type

Agent type assigned to the concrete generator during loading.
compatible_listener_types

Listener types assigned during loading that can create this generator's agents.
creating_agent_template

Agent template that created this generator, assigned during loading.

Create a new agent generator instance with the given name, description, and parameters.

Parameters:

Name Type Description Default
name str | None

Human-readable label for this generator run, used in log messages and serialized output. When None a random human-readable name is generated. The name is display metadata that is independent of parameters; it is never derived from them.

 None
description str

Optional longer description of what this particular run produces.

 ''
parameters dict[str, Any] | None

Key-value configuration values passed to each build step. Must satisfy the options declared by the owning agent template.

 None

agent_generator_build_steps = [(agent_generator_build_step(agent_templates_payload_service=(self.agent_templates_payload_service))) for agent_generator_build_step in (self.__class__.agent_generator_build_steps)] class-attribute instance-attribute

name = name instance-attribute

description = description instance-attribute

parameters = parameters instance-attribute

agent_generator_id = uuid.uuid4() instance-attribute

agent_templates_payload_service = AgentTemplatesPayloadsService(agent_template_id=(self.creating_agent_template.agent_template_id)) instance-attribute

datetime_created = utc_now() instance-attribute

environment = types.SimpleNamespace() instance-attribute

logger = logger.bind(logger_name=f'Agent Generator {self}', logger_type=(LoggerType.GENERATOR_LOGGER)) instance-attribute

event_logger = EventLogger(event_log=(EventLog(subject_id=(self.agent_generator_id))), system_logger=(self.logger)) instance-attribute

on_started() async

Hook invoked before build steps begin executing.

Override to perform any initialization that must complete before the build pipeline starts, such as preparing directories or acquiring external resources.

on_completed() async

Hook invoked after all build steps finish successfully.

Override to perform cleanup, notifications, or post-processing after a successful build.

on_running() async

Run build steps sequentially and report pipeline progress.

This method is final and must not be overridden.

Raises:

Type Description
ComponentRuntimeError

If a build step enters an errored or fatal state.

on_stopped() async

Hook invoked when the generator is stopped before all steps complete.

Override to clean up resources that were allocated before the generator was halted.

on_cancelled() async

Hook invoked when the generator run is cancelled externally.

Override to clean up resources when the build is aborted mid-pipeline.

on_errored(error) async

Hook invoked when a runtime error occurs during execution.

Parameters:

Name Type Description Default
error AgentGeneratorRuntimeError

The structured runtime error describing what failed and why.

 required

on_fatal(exc, phase) async

Hook invoked when an unrecoverable error occurs in the generator lifecycle.

Parameters:

Name Type Description Default
exc Exception

The underlying exception that triggered the fatal transition.

 required
phase ComponentLifeCyclePhase

The lifecycle phase (starting, running, stopping, etc.) during which the fatal error occurred.

 required

start() async

Start the agent generator and begin executing its build pipeline.

Raises:

Type Description
AgentGeneratorAlreadyRunningError

If the generator is already in a running state.
AgentGeneratorStartError

If the generator fails to start due to a lifecycle error.
AgentGeneratorFatalError

If an unhandled exception escapes on_started, leaving the generator in a fatal state. The original exception is chained onto it as __cause__.

cancel() async

Cancel the agent generator run.

Raises:

Type Description
AgentGeneratorNotRunningError

If the generator is not currently running.
AgentGeneratorFatalError

If an unhandled exception escapes on_cancelled, leaving the generator in a fatal state. The original exception is chained onto it as __cause__.

stop() async

Stop the generator and interrupt the currently executing build step.

Also attempts to stop the active build step if one is running.

Raises:

Type Description
AgentGeneratorNotRunningError

If the generator is not currently running.
AgentGeneratorStopError

If the generator fails to stop cleanly.
AgentGeneratorFatalError

If an unhandled exception escapes on_stopped, leaving the generator in a fatal state. The original exception is chained onto it as __cause__.

to_json(limit=10, offset=None, include_event_log_entries=True)

Serialize the generator's current state to a JSON-compatible dictionary.

Parameters:

Name Type Description Default
limit int

Maximum number of event log entries to include.

 10
offset int | None

Sequence offset to start the event log window from. If None, the tail (most recent entries up to limit) is returned.

 None
include_event_log_entries bool

When False, the event log's entries list is omitted (its total count and current progress are still included). Used by collection endpoints to keep list responses bounded.

 True

Returns:

Type Description
dict[str, JsonValue]

A dictionary containing the generator ID, name, description, parameters,
dict[str, JsonValue]

status, creation timestamp, event log, build step states, agent type,
dict[str, JsonValue]

compatible listener types, and a reference to the creating agent template.

to_json_reference()

Serialize a compact reference to this generator.

Returns:

Type Description
dict[str, JsonValue]

A dictionary containing only the generator ID, name, and agent type,
dict[str, JsonValue]

suitable for embedding as a lightweight foreign key reference in other
dict[str, JsonValue]

JSON objects.

BaseAgentGeneratorBuildStep(agent_templates_payload_service)

Bases: ComponentLifeCycle

A single step in an agent generator's build pipeline.

Subclasses implement build() to perform one discrete stage of the agent creation process (compilation, signing, packaging, uploading, etc.). Steps run sequentially within a BaseAgentGenerator and share a mutable SimpleNamespace environment so earlier steps can pass state (file paths, keys, metadata, etc.) to later ones.

Attributes:

Name Type Description
name str

Unique display name for this build step. Required.
description str

Human-readable explanation of what this step does.
datetime_started

UTC timestamp at which the most recent run started, or None.
datetime_stopped

UTC timestamp at which the most recent run stopped, or None.
environment

Shared mutable namespace passed between build steps in a pipeline.
parameters

Configuration parameters supplied by the owning agent generator.
logger

System logger for this build step.
event_logger EventLogger | None

Event logger shared with the owning agent generator, so events recorded by every build step appear together in the generator's single consolidated event log. Injected by the generator before the step runs; None until then.
agent_templates_payload_service

Service used to store and retrieve payload artifacts associated with the creating agent template.
root_directory

Directory containing the concrete build step's source file.
services

Server services exposed to the concrete build step.

Initialize the build step with a reference to the agent templates payload service.

Parameters:

Name Type Description Default
agent_templates_payload_service AgentTemplatesPayloadsService

Service providing storage and retrieval of payload artifacts produced or consumed by this build step.

 required

name instance-attribute

description = '' class-attribute instance-attribute

datetime_started = None instance-attribute

datetime_stopped = None instance-attribute

environment = types.SimpleNamespace() instance-attribute

parameters = {} instance-attribute

logger = logger.bind(logger_name=f'Agent Generator Build Step {self}', logger_type=(LoggerType.GENERATOR_LOGGER)) instance-attribute

event_logger = None instance-attribute

agent_templates_payload_service = agent_templates_payload_service instance-attribute

time_elapsed_in_seconds property

Wall-clock duration of the most recent run in seconds.

Returns:

Type Description
float | None

Elapsed seconds between start and stop, or None if the step has not
float | None

completed or was never started.

build(parameters) async

Execute the build logic for this step.

Override to implement the step's discrete unit of work. The shared environment namespace is accessible via self.environment, and agent_templates_payload_service is available for storing build artifacts.

Parameters:

Name Type Description Default
parameters dict

The generator's configuration parameters passed through from the owning BaseAgentGenerator instance.

 required

on_started() async

Record the start time for the current build-step run.

This method is final and must not be overridden.

on_running() async

Execute the concrete build implementation with the current parameters.

This method is final and must not be overridden.

on_completed() async

Record the stop time after the build step completes successfully.

This method is final and must not be overridden.

on_stopped() async

Record the stop time after the build step is stopped.

This method is final and must not be overridden.

on_cancelled() async

Record the stop time after the build step is cancelled.

This method is final and must not be overridden.

on_errored(error) async

Record the stop time and log a build-step runtime error.

This method is final and must not be overridden.

Parameters:

Name Type Description Default
error AgentGeneratorBuildStepRuntimeError

The runtime error raised while the build step was executing.

 required

on_fatal(exc, phase) async

Hook invoked when an unrecoverable error occurs in the build step lifecycle.

Parameters:

Name Type Description Default
exc Exception

The underlying exception that triggered the fatal transition.

 required
phase ComponentLifeCyclePhase

The lifecycle phase (starting, running, stopping, etc.) during which the fatal error occurred.

 required

run(parameters, environment, event_logger) async

Start this build step with the provided parameters and environment, blocking until done.

This is the entry point called by BaseAgentGenerator during pipeline execution. It injects the shared environment, parameter set, and event logger before starting the component lifecycle.

Parameters:

Name Type Description Default
parameters dict

Key-value configuration parameters forwarded from the generator.

 required
environment SimpleNamespace

Shared namespace that allows steps to read and write state across the pipeline.

 required
event_logger EventLogger

Event logger sharing the generator's event log, so events recorded by this step appear alongside those of every other step.

 required

reset()

Reset timing state and status so the step can be reused in a subsequent generator run.

to_json()

Serialize the build step's current state to a JSON-compatible dictionary.

Returns:

Type Description
dict[str, JsonValue]

A dictionary containing the step name, description, start and stop
dict[str, JsonValue]

timestamps, elapsed time in seconds, and current lifecycle status.

to_json_reference()

Serialize a compact reference to this build step.

Returns:

Type Description
dict[str, JsonValue]

A dictionary containing only the step name, suitable for embedding as a
dict[str, JsonValue]

lightweight reference in other JSON objects.

BaseAgentTemplate

Bases: ComponentMetadata

Abstract base class for agent templates that govern agent generator creation.

An agent template declares the configuration schema (options, validating function), the generator class, agent type, and compatible listener types for a family of agent generators. It validates and constructs BaseAgentGenerator instances from user-supplied parameters, filling in defaults and running cross-field validation before instantiation.

Attributes:

Name Type Description
agent_generator type[BaseAgentGenerator]

The generator class that this template instantiates when creating a new agent generator.
agent_type type[BaseAgentType] | str

The agent type that identifies which capabilities the generated agent supports. Either the agent type class, or the name of an agent type declared by another agent profile as a string reference to it. Both forms are resolved to a shared agent type instance once every agent profile is loaded, so this always reads back as a BaseAgentType instance at runtime.
compatible_listener_types set[str] | None

Names of listener types that agents generated from this template can connect through.
options set[Options] | None

Configuration options accepted when creating a generator from this template. Converted to a name-keyed dict at class definition time.
validating_function Callable[[dict[str, Options]], None] | None

Optional single-argument callable that validates the full set of resolved option values before generator creation.

agent_generator instance-attribute

agent_type instance-attribute

compatible_listener_types = None class-attribute instance-attribute

options = None class-attribute instance-attribute

validating_function = None class-attribute instance-attribute

create_agent_generator(name=None, description='', parameters=None)

Create an agent generator instance from this template using the given configuration.

Validates all supplied parameters against the declared options, fills in defaults for omitted optional options, runs the optional validating_function, then instantiates the agent generator.

Parameters:

Name Type Description Default
name str | None

Display name for the new generator. If None, a random human-readable name is generated. A generator's name is display metadata only: it is never derived from parameters, and updating parameters later never changes it.

 None
description str

Optional human-readable description for this generator run.

 ''
parameters dict[str, Primitive | PrimitiveCollection] | None

Option values that configure the generator, keyed by option name. Missing required options raise an error; missing optional options are filled with their declared default values.

 None

Returns:

Type Description
BaseAgentGenerator

A new BaseAgentGenerator instance configured with the provided parameters.

Raises:

Type Description
MissingRequiredAgentTemplateOptionError

If a required option is absent from parameters.
AgentTemplateOptionNotFoundError

If parameters contains an unknown option name.
AgentTemplateOptionValueValidationError

If an option value fails type or constraint validation.
AgentTemplateValidatingFunctionError

If the template validating function rejects the resolved options.

to_json()

Serialize the agent template's full metadata to a JSON-compatible dictionary.

Returns:

Type Description
dict[str, JsonValue]

A dictionary containing the template ID, label, name, description, version,
dict[str, JsonValue]

framework compatibility, authors, dependencies, agent type, compatible
dict[str, JsonValue]

listener types, options, and any validating function documentation.

to_json_reference()

Serialize a compact reference to this agent template.

Returns:

Type Description
dict[str, JsonValue]

A dictionary containing only the template ID, label, and name, suitable
dict[str, JsonValue]

for embedding as a lightweight foreign key reference in other JSON objects.

BaseAgentType

Defines the set of capabilities available to a specific category of agent.

An agent type groups related BaseAgentCapability classes under a shared name, allowing the framework to route incoming task commands to the correct capability implementation. Declare agent_capabilities at the class level; the framework converts the set into a name-keyed dictionary at class definition time for fast lookup during task dispatch.

Attributes:

Name Type Description
name str

Unique identifier for this agent type. Required and must be non-empty.
agent_capabilities set[type[BaseAgentCapability]] | None

The capability classes this agent type exposes. Declared as a set at the class level; converted to a name-keyed dict at class definition time.

name instance-attribute

agent_capabilities = None class-attribute instance-attribute

to_json()

Serialize the agent type and its capabilities to a JSON-compatible dictionary.

Returns:

Type Description
dict[str, JsonValue]

A dictionary containing the agent type name and a nested mapping of each
dict[str, JsonValue]

capability name to its serialized JSON representation.

AgentCapabilityLaunchError(message='An error occurred while launching the agent capability.', detail=None)

Bases: BaseSignalException

Raise this exception from on_launch to deliberately deny a task from starting and report failure to the operator.