cubepi.middleware

Middleware

class

source

Attributes

• hitl: HitlBinding | None

Methods

transform_context

transform_context(messages: list[Message], *, ctx: AgentContext, signal: asyncio.Event | None = None) -> list[Message]

source

convert_to_llm

convert_to_llm(messages: list[Message], *, ctx: AgentContext) -> list[Message]

source

before_tool_call

before_tool_call(ctx: BeforeToolCallContext, *, signal: asyncio.Event | None = None) -> BeforeToolCallResult | None

source

after_tool_call

after_tool_call(ctx: AfterToolCallContext, *, signal: asyncio.Event | None = None) -> AfterToolCallResult | None

source

transform_system_prompt

transform_system_prompt(system_prompt: str, *, ctx: AgentContext, signal: asyncio.Event | None = None) -> str

source

should_stop_after_turn

should_stop_after_turn(ctx: AgentContext) -> bool

source

after_model_response

after_model_response(response: AssistantMessage, ctx: AgentContext, *, signal: asyncio.Event | None = None) -> TurnAction | None

source

on_run_end

on_run_end(ctx: AgentContext, *, signal: asyncio.Event | None = None) -> list[Message] | None

source

extra_llm_calls

extra_llm_calls() -> Iterable[tuple[Provider, Model]]

Declare LLM calls this middleware drives outside the agent's main provider/model.

Each pair is (provider, model). cubepi.tracing.Recorder uses these to:

• Subscribe listeners on any provider the recorder isn't already watching, so the resulting calls show up in the trace tree alongside the agent's own chat spans.
• Identify middleware-owned calls by (model.provider, model.id) so they don't overwrite the root invoke_agent span's attribution (provider name, system prompt hash, tool list). This model-based gate is what handles the common "reuse one provider client, swap the model" pattern — listener identity alone would attribute the middleware's first call to the agent.

Default is empty — middlewares that do not call any LLM directly need not override.

source

TurnAction

class

TurnAction(self, response: AssistantMessage | None = None, inject_messages: list[Message] = list(), decision: Literal['natural', 'stop', 'loop_to_model'] = 'natural')

Directs the agent loop's next step after a model response.

Composition (chain): each middleware sees previous middleware's TurnAction. Last middleware's value wins for response and decision. inject_messages concatenates across the chain.

source

Attributes

• response: AssistantMessage | None
• inject_messages: list[Message]
• decision: Literal['natural', 'stop', 'loop_to_model']

compose_middleware

function

compose_middleware(middlewares: list[Middleware]) -> dict[str, Callable]

source

CompactionMiddleware

class

CompactionMiddleware(self, *, summary_model: BoundModel, max_tokens_before_compact: int, keep_recent_messages: int = 8, max_summary_tokens: int = 1024, min_compact_messages: int = 4)

Keep long histories within context by summarizing older turns.

source

Methods

transform_context

transform_context(messages: list[Message], *, ctx: AgentContext, signal: asyncio.Event | None = None) -> list[Message]

source

extra_llm_calls

extra_llm_calls() -> tuple[tuple[Provider, Model], ...]

source

CompactionState

class

JSON-safe summary state stored in AgentContext.extra.

source

Attributes

• summary: str
• summarized_message_ids: list[str]
• summarized_message_refs: list[str]
• last_summarized_message_id: str | None

SubagentMiddleware

class

SubagentMiddleware(self, *, subagents: dict[str, SubagentSpec], default_model: BoundModel, shared_tools: Sequence[AgentTool[BaseModel]] = (), inherited_middleware: Sequence[Middleware] = (), excluded_tool_names: set[str] | None = None, event_mapper: EventMapper | None = None, event_handler: EventHandler | None = None, tracer: SubagentTracer | None = None)

Inject a tool that delegates one task to an ephemeral child agent.

source

Attributes

• tools: list[AgentTool[BaseModel]]
• subagents: dict[str, SubagentSpec]
• shared_tools: tuple[AgentTool[BaseModel], ...]

SubagentRequest

class

source

Attributes

• name: str
• role: str
• task: str
• prompt: str
• subagent_type: str

SubagentResult

class

SubagentResult(self, agent_id: str, text: str, events: list[StructuredValue], error: str | None = None)

source

Attributes

• agent_id: str
• text: str
• events: list[StructuredValue]
• error: str | None

SubagentSpec

class

SubagentSpec(self, name: str, description: str, system_prompt: str, model: BoundModel | None = None, tools: Sequence[AgentTool[BaseModel]] = tuple(), middleware: Sequence[Middleware] = tuple())

source

Attributes

• name: str
• description: str
• system_prompt: str
• model: BoundModel | None
• tools: Sequence[AgentTool[BaseModel]]
• middleware: Sequence[Middleware]