Agentic Context Engine (ACE)

ACE (Agentic Context Engine) is a pipeline-based framework designed for building self-improving AI agents. It enables agents to learn from their own executions through a continuous loop of execution, reflection, and skill updating.

Core Components

• Pipeline Engine: Handles composition, concurrency, and typed step contracts using requires/provides logic.
• Roles: Business logic implementations (e.g., Agent, Reflector, SkillManager) backed by PydanticAI agents for structured LLM interactions.
• Skillbook: An evolving knowledge base of strategies that agents read from and learning loops write to.
• Steps: Context plumbing that extracts data, calls a role, and returns results to the pipeline.
• Runners: Orchestration components that compose steps into pipelines and manage iterations (e.g., ACERunner).
• Observability: Integrated via Logfire auto-instrumentation of PydanticAI agent calls.

ACE provides various runners and step-based integrations for different agentic frameworks. Integrations are categorized by their input type and the level of insight they provide (Micro, Meso, or Cloud).

The Skillbook is the ACE knowledge store, acting as a structured collection of learned issues and insights. It stores skills, which are individual entries representing a problem (issue) and a recommended action (insight).

Skills are categorized into two distinct sections:

• context: Learnings the agent should apply while solving a task.
• harness: Environment or runtime learnings that affect the pipeline itself.

Each skill includes metadata such as a unique id, keywords (for topic labeling), an active status, effectiveness counters (helpful_count, harmful_count, neutral_count), and occurrences for provenance tracing.

{
  "id": "context-00001",
  "section": "context",
  "keywords": ["math", "decomposition"],
  "issue": "Complex arithmetic questions are easier to solve when the work is decomposed into smaller verified steps.",
  "insight": "Break the problem into smaller steps before computing.",
  "active": true,
  "helpful_count": 5,
  "harmful_count": 0,
  "neutral_count": 1
}

The Pipeline Engine is a lightweight, domain-agnostic framework for composing processing steps into pipelines. It uses three core primitives to handle all composition patterns:

1. Sequential: Steps run one after another.
2. Branch: Forks the data to run multiple pipelines in parallel, then joins them.
3. Nesting: A Pipeline can be used as a single step within another pipeline.

Key features include:

• Contract Validation: Steps declare requires and provides fields. The pipeline validates that data dependencies are met at construction time, preventing runtime wiring errors.
• Immutable Context: Steps receive a frozen StepContext and must return a new one via .replace(), ensuring thread safety.
• Per-sample Error Isolation: A failure in one sample does not block others; every sample produces a SampleResult.
• Observation Hooks: Use PipelineHook to observe step transitions (e.g., for logging or metrics) without altering data flow.

A Branch is a specialized step that implements StepProtocol, allowing you to run multiple pipelines in parallel on the same input. It forks the context to all child pipelines, executes them concurrently, and then implicitly joins (merges) their outputs before the next step in the main pipeline begins.

Key characteristics:

• Parallelism: Branches run in parallel using a ThreadPoolExecutor (sync) or asyncio.gather (async).
• Immutability: All branches receive the same frozen context. Since StepContext is immutable, branches cannot corrupt each other's input.
• Contract Inference: A Branch automatically computes its requires and provides by taking the union of its children's requirements and provisions. This allows for seamless nesting of branches within pipelines.

graph LR
    T[Tokenize] --> U[Uppercase]
    T --> R[Reverse]
    U --> S[Summarize]
    R --> S

Session Isolation

Every tool call is scoped to a session_id. The server lazily initializes session runners (defaulting to ACELiteLLM) and manages their lifecycle using a configurable TTL (session_ttl_seconds).

Safety Modes

The server provides two layers of protection for non-production or read-only environments:

1. Safe Mode (ACE_MCP_SAFE_MODE=true): Blocks all mutating operations, including learning and skillbook persistence. Attempting these will return ACE_MCP_FORBIDDEN_IN_SAFE_MODE.
2. Save/Load Restriction (ACE_MCP_ALLOW_SAVE_LOAD=false): Specifically disables ace.skillbook.save and ace.skillbook.load. Attempting these will return ACE_MCP_SAVE_LOAD_DISABLED.

Path Security

When using ace.skillbook.save or ace.skillbook.load, the server resolves the provided path to an absolute canonical path. If ACE_MCP_SKILLBOOK_ROOT is set, the server will reject any paths that resolve to locations outside of that root directory.

A step in the ACE pipeline is a class that must implement three specific attributes and a __call__ method:

1. requires: A frozenset of keys that must exist in the StepContext.metadata before this step runs.
2. provides: A frozenset of keys that this step will add to the StepContext.metadata.
3. __call__(self, ctx: StepContext) -> StepContext: The execution logic. It should return a new StepContext (typically using ctx.replace(metadata=...)) containing the updated metadata.

Note: Use types.MappingProxyType when updating metadata to ensure immutability/correctness where required.

from types import MappingProxyType
from pipeline import StepContext

class Tokenize:
    """Split text into tokens and count words."""
    requires = frozenset()
    provides = frozenset({"tokens", "word_count"})

    def __call__(self, ctx: StepContext) -> StepContext:
        tokens = str(ctx.sample).split()
        return ctx.replace(
            metadata=MappingProxyType({
                **ctx.metadata,
                "tokens": tokens,
                "word_count": len(tokens),
            })
        )

To prevent context window exhaustion, the system employs a two-tier compaction strategy when PydanticAI: UsageLimitExceeded is triggered.

Tier 1: microcompact

• Action: Clears old execute_code tool results but keeps the last 3 tool results intact.
• Retention: Keeps all model messages (the reasoning chain).
• Flow: If the context size is still too large after microcompaction, the system falls through to Tier 2.

Tier 2: summarize_and_compact

• Action: Uses the LLM to summarize the progress (consumes 1 request from the budget).
• Context Preservation: Saves the pre-compaction context to the sandbox history variable.
• Replacement: Replaces the history with a [summary + continuation prompt].
• Limit: Compaction is capped at max_compactions=3.

Compaction Callback

RecursiveAgent.on_compaction() is used to save compaction metadata to the sandbox's history variable, allowing the agent to reference prior context even after it has been summarized.

StepContext is an immutable, frozen dataclass that carries data (sample and metadata) between steps.

Immutability & The .replace() Pattern Steps must never mutate the incoming context. Instead, they must use the .replace() method to return a new instance of the context with updated values. This ensures thread safety, branch safety, and a clear trace of data transformations.

Metadata vs. Named Fields

• metadata: Use a MappingProxyType (automatically coerced if you pass a plain dict) for transient, step-specific, or integration-specific data (e.g., metadata["debug_log"]).
• Named Fields: Subclass StepContext to add named, type-checkable fields for data shared across multiple steps in a pipeline (e.g., predictions, scores).

@dataclass(frozen=True)
class MLContext(StepContext):
    # Shared configuration
    model_config: dict | None = None

    # Produced by steps (None until the providing step runs)
    predictions: list | None = None
    scores: dict | None = None
    report: str | None = None