Agent API Reference¶

This is the complete API reference for the two agent profiles and the Mesh orchestrator. All signatures, attribute names, parameter types, and return shapes are sourced from the framework source code.

AutoAgent¶

from jarviscore import AutoAgent

AutoAgent is the framework-managed execution profile. Three class attributes form the minimum valid declaration. Production agents can additionally describe capabilities, declare provider authority, validate output, and select execution behavior.

Required class attributes¶

Attribute	Type	Description
`role`	`str`	Agent role identifier used for identity, profile loading, and explicit role routing. Example: `"researcher"`
`capabilities`	`List[str]`	List of capability strings this agent provides. Used by the Mesh workflow engine for task routing. Example: `["research", "analysis"]`
`system_prompt`	`str`	System prompt prepended to every Kernel call. Omitting this raises `ValueError` at instantiation.

Optional class attributes¶

Attribute	Type	Default	Description
`description`	`str`	`""` when absent	Human-readable role purpose used as fallback routing context.
`capability_descriptions`	`Dict[str, str]`	`{}`	Routing description for each capability. Distributed planning uses these descriptions instead of inferring intent from tags alone.
`capability_contracts`	`Dict[str, dict]`	`{}` when absent	Authorized `effects`, provider `systems`, and planner-visible artifact production and consumption declarations.
`artifact_reference_paths`	`tuple[tuple[str, ...], ...]`	`()`	Output paths whose non-null artifacts must be exact `artifact_ref` values hydrated from dependency outputs before validation. Use `"*"` for list members.
`output_schema`	`type[BaseModel]`	`None`	Optional Pydantic model enforced on CoderSubAgent execution output. Other role outputs require application validation.
`goal_oriented`	`bool`	`False`	When `True`, tasks are classified first: complex work uses `Plan → Execute → Evaluate`; bounded work can run as one direct Kernel turn. See Planning.
`default_kernel_role`	`str`	`None`	Fallback sub-agent role when the Planner emits `subagent_hint: null`. Built-in values are `"coder"`, `"researcher"`, `"communicator"`, and `"browser"`; products may register custom roles through an extended Kernel. Leave `None` for generalist agents.
`requires_auth`	`bool`	`False`	Opts into post-`setup()` `AuthenticationManager` injection when connected-app authentication is configured. Connected-app calls require a reachable Nexus Gateway.

capability_contracts uses this shape:

capability_contracts = {
    "code_review": {
        "effects": ["read", "propose"],
        "systems": ["github"],
        "produces": "ReviewReport(findings, inspected_revision)",
        "artifact_types": ["ReviewReport"],
        "requires_artifact_types": ["RepositorySnapshot"],
    },
}

Effects are read, propose, write, notify, or destructive. effects and systems describe authority; they do not grant credentials or bypass provider policy. produces is descriptive planning metadata. artifact_types and requires_artifact_types are optional stable names used by the DAG compiler to guarantee direct artifact-delivery dependencies. They grant no authority and perform no domain validation.

Optional environment overrides¶

Variable	Default	Description
`HITL_ENABLED`	`false`	Enable typed human-only HITL. Routine failure, low confidence and token spend never escalate.
`BROWSER_ENABLED`	`false`	Activate `BrowserSubAgent` for web automation tasks.
`MAX_GOAL_STEPS`	`30`	Hard step ceiling for goal-oriented agents.
`MAX_REPLAN_ATTEMPTS`	`8`	Maximum replanning cycles before the goal is marked failed.

execute_task¶

async def execute_task(task: Dict[str, Any]) -> Dict[str, Any]

The primary task entry point. The Mesh workflow engine calls it, and application code may call it directly after the agent has been started by a Mesh.

Input:

Key	Type	Description
`task`	`str`	Natural language task description
`context`	`dict`	Optional context forwarded to the Kernel

Return value (standard Kernel path):

Key	Type	Description
`status`	`str`	`"success"`, `"failure"`, `"yield"`, or `"hitl"` for a paused goal-oriented execution
`output`	`Any`	Task result payload
`payload`	`Any`	Alias of `output` on the standard Kernel path
`result_summary`	`str`	Guaranteed plain-prose display summary; structured data remains in `output`, `payload`, or `goal_execution`
`error`	`str \\| None`	`None` on success; failure or yield explanation otherwise
`tokens`	`dict`	Token usage: `{"input": int, "output": int, "total": int}`
`cost_usd`	`float`	Estimated cost in USD
`repairs`	`int`	Autonomous repair attempts. The Kernel path reports `0`; the legacy path reports attempts performed.
`agent_id`	`str`	The agent's unique identifier
`role`	`str`	The agent's role
`function_id`	`str \\| None`	FunctionRegistry atom ID if the task was registered
`dispatches`	`list`	Sub-agent dispatch log from the Kernel
`yield_metadata`	`dict`	Typed continuation or HITL metadata when execution yields; otherwise empty
`result_id`	`str`	Present when a configured ResultHandler stores the result

Additional keys when goal_oriented = True:

Key	Type	Description
`goal_execution`	`dict`	Planning summary for complex work, or direct-Kernel classification metadata for bounded work

Return value (legacy fallback path):

The legacy pipeline is used when the Kernel has not been initialized. It adds:

Key	Type	Description
`code`	`str`	The generated code that was executed

repairs remains part of the common envelope and may be greater than zero on this path.

setup¶

async def setup() -> None

Called by the Mesh before the agent receives any tasks. Initialises the LLM client, search client, code generator, sandbox executor, autonomous repair system, result handler, function registry, and Kernel. Override to add custom initialisation, but always call await super().setup() first.

teardown¶

async def teardown() -> None

Called by the Mesh on shutdown. Override to release resources such as database connections or open file handles.

CustomAgent¶

from jarviscore import CustomAgent

CustomAgent is the user-controlled execution profile. You own the execution logic entirely by implementing on_peer_request(). The framework provides P2P message routing, FastAPI lifecycle integration, and Mesh registration.

Required class attributes¶

Attribute	Type	Description
`role`	`str`	Agent role identifier
`capabilities`	`List[str]`	List of capability strings

Configuration attributes¶

Attribute	Type	Default	Description
`listen_timeout`	`float`	`1.0`	Seconds to wait for a P2P message before looping. Allows periodic `shutdown_requested` checks.
`auto_respond`	`bool`	`True`	When `True`, the return value of `on_peer_request()` is automatically sent as the response to the caller. Set to `False` to manage responses manually.

on_peer_request¶

async def on_peer_request(msg) -> Any

Primary handler for request-response P2P messages. Override this to implement your agent's logic.

msg attributes:

Attribute	Type	Description
`msg.sender`	`str`	Sender agent ID or role
`msg.data`	`dict`	Request payload
`msg.correlation_id`	`str`	Used by the framework for response matching; handled automatically

Return any value. When auto_respond = True, the return value is sent back to the sender automatically. Return None to skip the automatic response.

on_peer_notify¶

async def on_peer_notify(msg) -> None

Handler for fire-and-forget P2P notifications. No response is sent. Override to process events from other agents.

on_error¶

async def on_error(error: Exception, msg=None) -> None

Called when message processing raises an exception. Default implementation logs the error and continues. Override to add alerting, error tracking, or custom recovery logic.

execute_task¶

async def execute_task(task: Dict[str, Any]) -> Dict[str, Any]

Called by the Mesh workflow engine. The default implementation wraps the task dict in a synthetic IncomingMessage and calls on_peer_request(). Override this directly if you need different workflow integration behaviour.

Raises NotImplementedError if on_peer_request() returns None and execute_task() has not been overridden.

run¶

async def run() -> None

The P2P listener loop. Runs continuously in the background when the agent is started via JarvisLifespan. Dispatches incoming messages to on_peer_request() or on_peer_notify() based on message type. You do not need to override this.

Mesh¶

from jarviscore import Mesh

The runtime host for agent lifecycle, workflow execution and infrastructure detection. Distributed goal work is peer-claimed; Mesh does not assign a master agent or retain planning authority.

Constructor¶

Mesh(config: Optional[Dict[str, Any]] = None)

Config key	Type	Default	Description
`redis_url`	`str`	from `REDIS_URL` env	Redis connection string
`p2p_enabled`	`bool`	from `P2P_ENABLED` env	Enable SWIM/ZMQ peer transport
`checkpoint_interval`	`int`	`1`	Save workflow checkpoints every N steps
`max_parallel`	`int`	`5`	Maximum parallel step execution
`distributed_poll_interval`	`float`	`2.0`	Redis DAG polling interval in seconds
`distributed_claim_lease_seconds`	`int`	`60`	Renewable execution-claim lease duration
`mesh_planning_lease_seconds`	`int`	`300`	Goal compilation or amendment lease duration
`mesh_max_reconciliation_revisions`	`int`	`3`	Maximum bounded semantic reconciliation revision
`mesh_response_capability`	`str`	`None`	Capability that owns the optional `final_response` step
`execution_budget`	`dict`	`ExecutionBudget` defaults	Shared limits for the complete distributed goal

The Mesh auto-detects available infrastructure at start() time. Do not pass mode=: that argument is deprecated and has no effect.

Methods¶

add¶

def add(agent_class_or_instance, agent_id: Optional[str] = None, **kwargs) -> Agent

Register an agent with the Mesh. Accepts a class (which will be instantiated) or a pre-instantiated agent. Returns the Agent instance.

Raises ValueError if an agent with the same agent_id is already registered, or if a class-based agent with the same role is added without an explicit agent_id. Raises TypeError if the class does not inherit from Agent.

start¶

async def start() -> None

Probe infrastructure; inject stores, mailbox, and HITL; call setup() on every agent; attach peer clients and optional authentication; then start the workflow engine. Must be called before task execution.

Raises RuntimeError if no agents are registered or if start() has already been called.

workflow¶

async def workflow(
    workflow_id: str,
    steps: List[Dict[str, Any]],
    timeout_per_step: Optional[float] = None,
) -> List[Dict[str, Any]]

Execute a multi-step workflow. Returns a list of step results in execution order.

Each step dict:

Key	Type	Required	Description
`id`	`str`	No	Stable step ID. Generated when omitted. Prefer explicit IDs for readable dependencies and recovery.
`agent`	`str`	Yes	Agent role or capability that should execute this step
`task`	`str`	Yes	Natural language task description
`depends_on`	`List[str \\| int]`	No	Explicit step IDs or legacy zero-based indices this step depends on
`context`	`dict`	No	Additional context passed to `execute_task()`
`complexity`	`str`	No	Model tier hint: `"nano"`, `"standard"`, or `"heavy"`
`timeout`	`float`	No	Per-step timeout overriding `timeout_per_step` and `WORKFLOW_STEP_TIMEOUT`

Raises RuntimeError if start() has not been called or if the workflow engine is unavailable.

execute_goal¶

async def execute_goal(
    goal: str,
    *,
    workflow_id: Optional[str] = None,
    context: Optional[Dict[str, Any]] = None,
    timeout: Optional[float] = None,
) -> Dict[str, Any]

Register an immutable source goal, wait for any planning-capable node to publish a capability-addressed Redis DAG, and observe independent peer claims until the workflow reaches completed, failed, or waiting. Requires Redis.

Set config["execution_budget"] on Mesh to bound the complete execution:

mesh = Mesh(config={
    "execution_budget": {
        "max_seconds": 900,
        "max_tokens": 240_000,
        "max_steps": 30,
        "max_replans": 8,
        "max_peer_depth": 2,
        "peer_timeout_seconds": 300,
    }
})

The framework stores this as ExecutionBudget in the durable WorkflowEnvelope. Caller context cannot override it. Token usage is enforced through one Redis-backed account per workflow: each model call reserves capacity before dispatch and settles exact provider-reported usage and cost afterward. Planning, evaluators, dependency/effect reviews, direct steps and nested peer mandates all debit that same account.

Atom repair lifecycle¶

Coder exposes inspect_atom_for_repair and repair_atom only after an atom has failed during the current run. A repair candidate is bound to the atom's name, provider, version and failed invocation. register_function rejects candidates without successful execution evidence and rejects stale or renamed repairs. A successful repair becomes the next immutable FunctionRegistry version while the superseded source remains available for audit.

Distributed execution envelopes¶

from jarviscore.orchestration import (
    CapabilityMandate,
    ExecutionBudget,
    WorkflowEnvelope,
    WorkflowEvidence,
)

WorkflowEnvelope is the canonical source/context/DAG record. CapabilityMandate is the scoped peer-request lifecycle. WorkflowEvidence is the complete artifact/interpretation/state snapshot used for final synthesis, including the current obligation projection. These types serialize compatibly with existing Redis records.

Mesh.execute_goal() results distinguish process lifecycle from source-goal truth:

status reports whether the distributed execution completed, failed, waited or was cancelled.
obligation_status reports satisfied, blocked or incomplete.
response_status reports completed, failed, waiting or not_required for the current revision's user-facing response.

The remaining result fields are:

Field	Description
`workflow_id`	Durable caller-supplied or generated identity
`goal`	Exact source goal
`obligations`	Independently verifiable source requirements
`revision`	Current published plan revision
`result_summary`	Current revision's terminal user response, when present
`steps`	Immutable attempts from every revision, each carrying `plan_revision`

A terminal step with an actionable semantic gap can trigger a bounded DAG revision. Reconciliation appends new work only for unresolved obligation IDs, preserves satisfied obligations and completed effects, records supersession lineage, and emits a fresh final response. If no available capability can advance the gap, the current revision settles as blocked rather than looping or claiming that execution completion satisfied the goal.

status, obligation_status and response_status are independent. A response failure can therefore coexist with satisfied business obligations. See Durable Goal Execution.

resume_goal¶

async def resume_goal(
    workflow_id: str,
    step_id: str,
    *,
    context: Optional[Dict[str, Any]] = None,
    timeout: float = 900.0,
) -> Dict[str, Any]

Resume a waiting step with optional human or external context. The original executor affinity is retained and blocked descendants become claimable after the resumed step succeeds.

cancel_goal¶

def cancel_goal(
    workflow_id: str,
    *,
    reason: str = "Goal cancelled",
) -> bool

Durably cancel shared work. Cancellation fences later claims and terminal writes, including writes from an executor whose lease expired before cancellation. Requires Redis. Cancelling the coroutine running execute_goal() performs the same durable cancellation before re-raising CancelledError.

replan_goal¶

async def replan_goal(
    workflow_id: str,
    *,
    reason: str,
    context: Optional[Dict[str, Any]] = None,
    timeout: float = 900.0,
) -> Dict[str, Any]

Compile an append-only delta for currently unresolved obligations under a short planning lease. The commit uses revision compare-and-swap, assigns new plan_revision values, and cannot remove, reuse or mutate earlier step IDs and outputs. Only obligations covered by the delta gain supersession lineage; other satisfied obligations keep their authoritative attempts. The source goal and obligation ledger remain immutable, and peers resume capability-based claiming after publication.

stop¶

async def stop() -> None

Request shutdown for all agents, call teardown() on each, stop the P2P coordinator and workflow engine, and close all infrastructure connections.

has_capability¶

def has_capability(cap: str) -> bool

Check whether an infrastructure capability is active. Capabilities are populated at start() time.

Value	Active when
`"workflow"`	Always (workflow engine always starts)
`"peer_local"`	Always (in-process routing always available)
`"redis"`	`REDIS_URL` is set and reachable
`"peer_distributed"`	Redis is active
`"peer_swim"`	`P2P_ENABLED=true` and SWIM coordinator started
`"blob"`	BlobStorage initialised (local backend always active)
`"nexus"`	NexusLocalStore initialised (always active, zero dep)
`"athena"`	`ATHENA_URL` is set and reachable
`"auth"`	`auth_mode` is set in the Mesh config
`"prometheus"`	`PROMETHEUS_ENABLED=true` and metrics server started

serve_forever¶

async def serve_forever() -> None

Block indefinitely, processing incoming tasks from the P2P network. For distributed mode deployments. Handles graceful shutdown on KeyboardInterrupt.

JarvisLifespan¶

from jarviscore.integrations.fastapi import JarvisLifespan

FastAPI lifespan manager for JarvisCore agents. Handles Mesh startup, background task management, graceful shutdown, and state injection.

Constructor¶

JarvisLifespan(
    agents: Union[Agent, List[Agent]],
    **mesh_config
)

Parameter	Type	Description
`agents`	`Agent \\| List[Agent]`	Single agent or list of agents to register with the Mesh
`**mesh_config`	`Any`	Forwarded to `Mesh(config=mesh_config)`. Common keys: `redis_url`, `p2p_enabled`, `bind_port`

On startup, injects into app.state:

Key	Value
`app.state.jarvis_mesh`	The `Mesh` instance
`app.state.jarvis_agents`	`Dict[str, Agent]` mapping `role → agent`

create_jarvis_app¶

from jarviscore.integrations.fastapi import create_jarvis_app

def create_jarvis_app(
    agent: Agent,
    title: str = "JarvisCore Agent",
    description: str = "API powered by JarvisCore",
    version: str = "1.0.0",
    **mesh_config
) -> FastAPI

Convenience wrapper for single-agent deployments. Creates a FastAPI app with JarvisLifespan pre-configured. The Mesh auto-detects its operational mode from available infrastructure at startup: no mode argument is accepted. For multi-agent deployments or more control, use JarvisLifespan directly.

Agent API Reference¶

AutoAgent¶

Required class attributes¶

Optional class attributes¶

Optional environment overrides¶

execute_task¶

setup¶

teardown¶

CustomAgent¶

Required class attributes¶

Configuration attributes¶

on_peer_request¶

on_peer_notify¶

on_error¶

execute_task¶

run¶

Mesh¶

Constructor¶

Methods¶

add¶

start¶

workflow¶

execute_goal¶

Atom repair lifecycle¶

Distributed execution envelopes¶

resume_goal¶

cancel_goal¶

replan_goal¶

stop¶

has_capability¶

serve_forever¶

JarvisLifespan¶

Constructor¶

create_jarvis_app¶

Further Reading¶