orrery-core¶

Shared library providing the foundation for all agents: agent factory, plugins, RBAC, guardrails, input validation, audit logging, resilience, and typed configuration.

Tool Results¶

ToolResult¶

A Pydantic model for structured tool outputs. While agents can return raw dicts, using ToolResult ensures consistency and enables machine-readable error handling for cross-agent composition (e.g., remediation loops).

from orrery_core import ToolResult


async def my_tool(name: str) -> dict:
    if not name:
        return ToolResult.error("Name is required", error_type="MissingInput").to_dict()

    return ToolResult.ok(message=f"Hello {name}", length=len(name)).to_dict()

• Factories: ok(), error(), partial().
• Fields: status, message, error_type, data, remediation_hints.
• Flattening: .to_dict() flattens data into the top level for backward compatibility with legacy consumers while keeping reserved fields (status, message, etc.) safe.

Operator Registry¶

OperatorRegistry¶

A central registry for Kubernetes operator awareness. It allows agents to reason about high-level custom resources (like Strimzi Kafka or ECK Elasticsearch) instead of just low-level Pods.

from orrery_core import default_registry

# Look up a detector by CRD group
detector = default_registry.get_by_group("kafka.strimzi.io")

# Interpret raw CR status into a unified OperatorStatus
status = detector.interpret_status("Kafka", raw_cr_dict)
print(status.healthy, status.phase, status.warnings)

Ships with built-in support for Strimzi and ECK. New operators can be registered by implementing the OperatorDetector protocol.

Plugins¶

Cross-cutting concerns are packaged as ADK BasePlugin subclasses and registered once on the Runner via default_plugins(). Plugins apply globally to every agent, tool, and LLM call — no per-agent callback wiring needed.

default_plugins()¶

Creates the standard set of plugins in the correct registration order:

from orrery_core import default_plugins
from google.adk.runners import Runner

runner = Runner(
    agent=root_agent,
    app_name="myapp",
    session_service=session_service,
    plugins=default_plugins(),
)

Each plugin's ADK adapter lives in orrery_core/plugins/<name>_plugin.py; the mechanism it wraps lives under the matching aspect package (e.g. ResiliencePlugin → reliability/resilience.py's CircuitBreaker).

Parameters:

Individual plugins¶

Each plugin can also be instantiated independently for custom configurations:

from orrery_core import (
    GuardrailsPlugin,
    ResiliencePlugin,
    MetricsPlugin,
    AuditPlugin,
    ActivityPlugin,
    ErrorHandlerPlugin,
)

plugins = [
    GuardrailsPlugin(mode="dry_run"),  # dry-run mode for testing
    ResiliencePlugin(failure_threshold=3),  # custom threshold
    MetricsPlugin(),
    AuditPlugin(log_path="audit.jsonl"),  # also write to local file
    ErrorHandlerPlugin(),
]

Plugins vs callbacks¶

Plugins registered on the Runner apply globally and run before any agent-level callbacks. If a plugin returns a value, the agent-level callback is skipped. Use agent-level callbacks only for agent-specific logic (e.g., Slack's custom confirmation buttons).

Agent Factory¶

create_agent()¶

Creates an ADK Agent with sensible defaults. Since cross-cutting concerns are handled by plugins, agent definitions are simple:

from orrery_core import create_agent, load_agent_env

load_agent_env(__file__)

root_agent = create_agent(
    name="my_agent",
    description="What this agent does.",
    instruction="How the agent should behave.",
    tools=[my_tool, another_tool],
)

Agent-level callback parameters (before_tool_callback, after_tool_callback, etc.) are still supported for agent-specific logic but are not needed for standard cross-cutting concerns.

resolve_planner()¶

Returns an ADK BasePlanner (or None) based on the ORRERY_PLANNER env var. Planning adds an explicit reasoning step before tool calls — useful for orchestration agents that route across specialists, less useful for narrow tool-leaf agents.

from orrery_core import create_agent, resolve_planner

_planner = resolve_planner()  # read once at import time

root_agent = create_agent(
    name="orchestrator",
    description="Routes across specialist agents.",
    instruction="…",
    tools=[…],
    planner=_planner,  # opt-in per agent
)

Builtin-only knobs (consulted only when ORRERY_PLANNER=builtin):

Where it's wired in orrery-assistant: the root orchestrator, the triage_summarizer, and the remediation_actor. All three are reasoning-heavy. Tool-leaf agents (per-system health checkers, the verifier, etc.) intentionally skip the planner — they would only add latency without improving output quality.

Deterministic multi-step flows — ADK 2.0 graph Workflow¶

The create_sequential_agent() / create_parallel_agent() / create_loop_agent() factories were removed in the ADK 2.0 upgrade — those SequentialAgent / ParallelAgent / LoopAgent wrappers are deprecated. Compose structured, non-LLM-delegated flows as a graph Workflow instead: create_agent() LlmAgents are graph nodes (an LlmAgent is a BaseNode in ADK 2.0), and routing is pure-Python FunctionNodes. Edges are chain-tuples (sequential), node-tuples (parallel), and RoutingMap dicts (conditional/loop); a JoinNode is the parallel fan-in barrier.

from google.adk import Workflow
from google.adk.workflow import JoinNode

health_join = JoinNode(name="health_join")

triage = Workflow(
    name="incident_triage",
    description="Full incident triage pipeline.",
    edges=[
        # Fan out to parallel checkers, join, then summarize → journal
        ("START", (kafka_checker, k8s_checker, docker_checker)),
        ((kafka_checker, k8s_checker, docker_checker), health_join),
        (health_join, summarizer, journal_writer),
    ],
)

Nodes pass data via output_key, which writes to session state for downstream nodes to read. See ADR-003 for the full rationale and the closed-loop remediation pattern (actor → verifier → verify_route bounded by a state counter, replacing LoopAgent).

load_agent_env(__file__)¶

Loads the .env file located next to the calling module.

Async Tools¶

All tool functions must be async def. Use asyncio.to_thread() or asyncio.create_subprocess_exec() to offload blocking I/O:

import asyncio
from orrery_core import with_retry, destructive


@with_retry(max_retries=3, retryable=(ConnectionError, TimeoutError))
async def list_topics() -> dict:
    admin = _get_admin_client()
    metadata = await asyncio.to_thread(admin.list_topics, timeout=10)
    return {"status": "success", "topics": list(metadata.topics.keys())}


@destructive("permanently deletes the topic")
async def delete_topic(topic_name: str) -> dict: ...

The @with_retry decorator automatically detects async functions and uses await asyncio.sleep() for backoff delays.

Resilience¶

Fault-tolerance utilities for handling transient failures and service outages.

ResiliencePlugin / CircuitBreaker¶

Per-tool circuit breaker applied globally via the ResiliencePlugin. When a tool exceeds the failure threshold, subsequent calls are short-circuited until a recovery timeout allows a probe.

@with_retry¶

Decorator for async tool functions that adds retry with exponential backoff and jitter.

from orrery_core import with_retry


@with_retry(max_retries=3, retryable=(ConnectionError, TimeoutError))
async def list_kafka_topics(timeout: int = 10) -> dict: ...

Apply @with_retry to read-only tools that call external services. Do not apply to mutating operations (create, delete) where retries could be dangerous.

Error Handling¶

ErrorHandlerPlugin¶

Handles both tool and model errors globally. When a tool raises an exception, it returns a structured error dict so the LLM can reason about it:

{"status": "error", "error_type": "KafkaException", "message": "Tool 'get_kafka_cluster_health' failed: broker down"}

When a model call fails, it returns a friendly LlmResponse message.

The underlying factories (graceful_tool_error(), graceful_model_error()) can still be used directly as agent-level callbacks if needed.

Input Validation¶

Reusable validators for tool inputs. Each returns None on success or {"status": "error", "message": ...} on failure. Use the walrus operator for concise early-return:

from orrery_core.validation import validate_string, validate_positive_int, KAFKA_TOPIC_PATTERN


async def create_kafka_topic(topic_name: str, num_partitions: int = 1) -> dict:
    if err := validate_string(topic_name, "topic_name", pattern=KAFKA_TOPIC_PATTERN):
        return err
    if err := validate_positive_int(num_partitions, "num_partitions", max_value=10_000):
        return err
    ...

Available Validators¶

Constants¶

Guardrails¶

Guardrails prevent destructive tools from executing without confirmation. They are enforced globally via the GuardrailsPlugin.

Marking tools as destructive¶

from orrery_core import destructive


@destructive("permanently deletes the topic and all its data")
async def delete_kafka_topic(topic_name: str) -> dict: ...

The @destructive() decorator marks the function with metadata. The GuardrailsPlugin reads this metadata at runtime and gates execution.

Confirmation flow¶

1. User asks to delete a topic
2. LLM calls delete_kafka_topic
3. GuardrailsPlugin intercepts → returns {"status": "confirmation_required", ...}
4. LLM receives the message and asks the user to confirm
5. User confirms → LLM calls the tool again → plugin allows it

The guardrail intercepts delete_kafka_topic (#12-#13), asks the user to confirm (#14), and only executes after explicit confirmation (#15-#18).

Guardrail modes¶

Role-Based Access Control (RBAC)¶

Three-role hierarchy that reuses existing guardrail metadata to control who can call which tools. See ADR-001 for the full design rationale.

VIEWER (0)   → read-only tools (unguarded)
OPERATOR (1) → + mutating tools (@confirm)
ADMIN (2)    → + destructive tools (@destructive)

RBAC is enforced globally by the GuardrailsPlugin. It runs before the confirmation gate — unauthorized users are blocked before reaching the confirmation prompt.

RolePolicy¶

Maps tools to their minimum required role. By default, roles are inferred from @destructive/@confirm decorators. Use overrides for exceptions:

from orrery_core import RolePolicy, Role, default_plugins

policy = RolePolicy(overrides={"sensitive_read": Role.OPERATOR})
plugins = default_plugins(role_policy=policy)

@requires_role()¶

Decorator for explicit role annotation on tools that don't use @destructive/@confirm:

from orrery_core import requires_role, Role


@requires_role(Role.ADMIN)
async def manage_users() -> dict: ...

Setting the user role¶

Use set_user_role() to assign a role from trusted server-side code:

from orrery_core import set_user_role

initial_state = {}
set_user_role(initial_state, "admin")
session = session_service.create_session(state=initial_state, ...)

Invalid role values default to viewer. The GuardrailsPlugin calls ensure_default_role() via before_agent_callback to prevent privilege escalation from untrusted session state.

Testing roles across surfaces¶

For a hands-on guide to exercising viewer / operator / admin from ADK Web, adk run, Slack, Google Chat, and a custom Runner, see Testing RBAC across surfaces.

Structured Logging¶

setup_logging()¶

Configures the root Python logger with a JSON formatter that outputs structured log lines to stdout — container-friendly and ready for Loki, ELK, Splunk, or Cloud Logging.

from orrery_core import setup_logging

setup_logging()  # call once at startup

Called automatically by load_agent_env(), so all agents get structured logging by default.

JSONFormatter¶

The underlying logging.Formatter that produces JSON output. Can be used standalone if you need custom handler configuration.

Audit Logging¶

AuditPlugin¶

Emits a structured audit entry for every tool invocation via the AuditPlugin. Each entry includes agent name, tool name, arguments, result, user ID, and session ID.

Sensitive arguments (containing password, secret, token, api_key, credential) are automatically redacted to ***.

An optional audit_log_path parameter in default_plugins() writes a local .jsonl file in addition to stdout.

Activity Tracking¶

ActivityPlugin¶

Records every tool call to session_log in session state via the ActivityPlugin. This makes all agent activity visible to get_session_summary() in the journal agent, regardless of which sub-agent performed the work.

Typed Configuration¶

AgentConfig¶

A pydantic-settings base class for typed, validated configuration. Replaces raw os.getenv() calls.

from orrery_core import AgentConfig, load_config


class KafkaConfig(AgentConfig):
    kafka_bootstrap_servers: str = "localhost:9092"


config = load_config(KafkaConfig, __file__)
print(config.kafka_bootstrap_servers)
print(config.model_provider)  # "gemini", "anthropic", "openai", etc.
print(config.model_name)  # "gemini-2.0-flash", "anthropic/claude-sonnet-4-20250514", etc.

Base fields (inherited by all configs):

resolve_model()¶

Resolves the LLM model from environment variables. Called automatically by create_agent() when no explicit model is passed.

• For MODEL_PROVIDER=gemini (default): returns a plain model string
• For any other provider: returns a LiteLlm instance via ADK's built-in LiteLLM integration

See Configuration reference for provider-specific env vars and API key setup.

load_config(ConfigClass, __file__)¶

Loads configuration from the .env file next to the calling module, with environment variable overrides.

Persistent Runner¶

run_persistent()¶

An async helper that runs an agent in a CLI loop with in-memory or PostgreSQL-backed sessions (set DATABASE_URL to persist). Supports plugins.

import asyncio
from orrery_core import run_persistent, default_plugins
from my_agent.agent import root_agent

asyncio.run(run_persistent(root_agent, app_name="my_agent", plugins=default_plugins()))

Session state (notes, preferences, bookmarks) survives across restarts. Type new for a fresh session or quit to exit.

For the web UI, use ADK's built-in persistence flag instead:

adk web --session_service_uri=postgresql+asyncpg://user:pass@host:5432/agents agents/my-agent