cubepi.middleware

Middleware

class

source

Methods

transform_context

transform_context(messages: list, *, signal = None) -> list

source

convert_to_llm

convert_to_llm(messages: list) -> list

source

before_tool_call

before_tool_call(ctx: Any, *, signal = None) -> Any

source

after_tool_call

after_tool_call(ctx: Any, *, signal = None) -> Any

source

transform_system_prompt

transform_system_prompt(system_prompt: str, *, signal = None) -> str

source

should_stop_after_turn

should_stop_after_turn(ctx: Any) -> bool

source

after_model_response

after_model_response(response: AssistantMessage, ctx: Any, *, signal: Any = None) -> TurnAction | 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