Skip to main content
Version: 0.7

cubepi.middleware

Middleware​

class

source

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

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

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]