API reference¶
Generated from the apogee-ai source with mkdocstrings. Every symbol below is exported from apogee_ai, so it is part of the supported public surface.
Other¶
Bases: Protocol
Records an audited action (adapter delegates to ai-audit's hash-chain).
async
¶
Python
record_tool(ctx: GovernanceContext, *, tool_name: str, status: str, detail: str = '') -> None
dataclass
¶
Python
Agent(name: str = '', slug: str = '', version: str = '0.1.0', description: str = '', status: AgentStatusEnum = DRAFT, visibility: AgentVisibilityEnum = PRIVATE, owner_id: str | None = None, tags: list[str] = list(), agent_type: AgentTypeEnum = WORKER, execution_pattern: ExecutionPatternEnum = REACT, collaboration_pattern: CollaborationPatternEnum = HIERARCHICAL, engine: EngineBinding = EngineBinding(), model: ModelConfig = ModelConfig(), prompt: PromptConfig = PromptConfig(), memory: MemoryConfig = MemoryConfig(), runtime: RuntimeConfig = RuntimeConfig(), tools: list[ToolBinding] = list(), managed_agents: list[str] = list(), parent_supervisor_id: str | None = None)
Bases: BaseEntity
Root aggregate representing a configurable AI agent.
Carries identity + the full versioned configuration. In a future
iteration we may split identity (Agent) from config (AgentVersion)
as discussed in the architectural plan; for the MVP they live together.
class-attribute
instance-attribute
¶
Python
execution_pattern: ExecutionPatternEnum = REACT
class-attribute
instance-attribute
¶
Python
collaboration_pattern: CollaborationPatternEnum = HIERARCHICAL
class-attribute
instance-attribute
¶
Python
engine: EngineBinding = field(default_factory=EngineBinding)
class-attribute
instance-attribute
¶
Python
model: ModelConfig = field(default_factory=ModelConfig)
class-attribute
instance-attribute
¶
Python
prompt: PromptConfig = field(default_factory=PromptConfig)
class-attribute
instance-attribute
¶
Python
memory: MemoryConfig = field(default_factory=MemoryConfig)
class-attribute
instance-attribute
¶
Python
runtime: RuntimeConfig = field(default_factory=RuntimeConfig)
class-attribute
instance-attribute
¶
Python
tools: list[ToolBinding] = field(default_factory=list)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
dataclass
¶
dataclass
¶
Python
AgentInput(user_input: str, history: list[Message] = list(), metadata: dict[str, Any] = dict())
dataclass
¶
Python
AgentOutput(final_response: str, messages: list[Message] = list(), iterations: int = 0, finish_reason: str = 'stop', metadata: dict[str, Any] = dict())
dataclass
¶
dataclass
¶
Python
ApprovalRequest(kind: str, payload: dict[str, Any] = dict(), metadata: dict[str, Any] = dict())
Snapshot of state the human is asked to approve.
Bases: Protocol
Pre-call gate + post-call charge (adapter delegates to ai-context-budget).
async
¶
Python
allow(ctx: GovernanceContext) -> BudgetVerdict
async
¶
Python
charge(ctx: GovernanceContext, *, prompt_tokens: int, completion_tokens: int) -> None
dataclass
¶
dataclass
¶
Python
CompiledAgent(agent: Agent, deps: EngineDependencies, native_object: Any = None)
dataclass
¶
Python
Conversation(agent_slug: str = '', user_id: str | None = None, title: str = '', messages: list[Message] = list(), closed: bool = False)
dataclass
¶
Python
Document(text: str, metadata: dict = dict(), id: str = (lambda: f'doc_{hex[:12]}')(), embedding: list[float] = list())
dataclass
¶
Python
EngineBinding(kind: EngineKindEnum = NATIVE, options: dict[str, Any] = dict())
class-attribute
instance-attribute
¶
dataclass
¶
Python
EngineDependencies(llm_client: ILLMClient, tool_executor: Any | None = None, extras: dict[str, Any] = dict())
class-attribute
instance-attribute
¶
dataclass
¶
dataclass
¶
dataclass
¶
Python
EvalReport(suite: str, results: list[EvalCaseResult] = list())
Source code in apogee_ai/application/eval/eval_runner.py
async
¶
Python
run(suite: EvalSuite, agent: Agent) -> EvalReport
Source code in apogee_ai/application/eval/eval_runner.py
Python
async def run(self, suite: EvalSuite, agent: Agent) -> EvalReport:
report = EvalReport(suite=suite.name)
for case in suite.cases:
output = await self._exec(agent, case.input)
failed: list[int] = []
for idx, matcher in enumerate(case.matchers):
if not matcher(output):
failed.append(idx)
report.results.append(
EvalCaseResult(
name=case.name,
passed=not failed,
output=output,
failed_matchers=failed,
)
)
return report
dataclass
¶
Python
ExecutionTrace(agent_slug: str = '', conversation_id: str | None = None, user_input: str = '', final_response: str = '', iterations: int = 0, finish_reason: str = 'stop', steps: list[TraceStep] = list(), tokens_prompt: int = 0, tokens_completion: int = 0, cost_usd: float = 0.0, metadata: dict[str, Any] = dict())
dataclass
¶
dataclass
¶
Python
GovernanceDeps(tool_guard: ToolGuardPort | None = None, audit: ActionAuditPort | None = None, budget: BudgetGuardPort | None = None, context: ContextManagerPort | None = None)
Optional governance hooks injected via EngineDependencies.extras['governance'].
Every field is optional; a None field disables that hook. An absent
GovernanceDeps (the default) disables all of them.
dataclass
¶
dataclass
¶
dataclass
¶
dataclass
¶
Python
KnowledgeSource(name: str = '', description: str = '', vector_store_id: str = '', embedding_model: str = 'text-embedding-3-small', metadata: dict = dict())
Bases: BaseEntity
class-attribute
instance-attribute
¶
dataclass
¶
Python
LLMRequest(messages: list[Message], model: ModelConfig, stop: list[str] = list(), tools: list[ToolSchema] = list())
class-attribute
instance-attribute
¶
Python
tools: list[ToolSchema] = field(default_factory=list)
dataclass
¶
Python
LLMResponse(content: str, finish_reason: str = 'stop', prompt_tokens: int = 0, completion_tokens: int = 0, raw: dict | None = None, tool_calls: list[ToolCall] = list())
dataclass
¶
Python
MemoryConfig(strategy: MemoryStrategyEnum = SLIDING_WINDOW, context_window: int = 8000, long_term: bool = False, store: str = 'in_memory', session_timeout: int = 3600)
dataclass
¶
Python
Message(role: RoleEnum, content: str, name: str | None = None, tool_calls: list[dict[str, Any]] = list(), created_at: datetime = (lambda: now(utc))())
dataclass
¶
Python
ModelConfig(provider: LLMProviderEnum = ECHO, model: str = 'echo', temperature: float = 0.7, max_tokens: int = 1024, top_p: float = 1.0, top_k: int | None = None, frequency_penalty: float = 0.0, presence_penalty: float = 0.0, stop_sequences: list[str] = list(), response_format: str = 'text', fallback_model: str | None = None)
class-attribute
instance-attribute
¶
dataclass
¶
Python
PromptConfig(system: str = 'You are a helpful assistant.', persona: str | None = None, language: str = 'pt-BR', greeting: str | None = None, template_ref: str | None = None, few_shot_examples: list[dict[str, str]] = list(), output_template: str | None = None)
class-attribute
instance-attribute
¶
dataclass
¶
Python
RuntimeConfig(max_iterations: int = 8, timeout_seconds: int = 60, streaming_enabled: bool = False, retry_attempts: int = 2, retry_backoff: str = 'exponential', fallback_agent_id: str | None = None)
dataclass
¶
Python
Tool(name: str = '', description: str = '', json_schema: dict[str, Any] = dict(), implementation_ref: str = '')
dataclass
¶
dataclass
¶
Bases: Protocol
Authorizes a tool invocation (adapter delegates to ai-authz).
async
¶
Python
authorize(ctx: GovernanceContext, *, tool_name: str) -> GuardVerdict
dataclass
¶
dataclass
¶
dataclass
¶
Python
TraceStep(kind: str, name: str, started_at: datetime = (lambda: now(utc))(), duration_ms: int = 0, input: dict[str, Any] = dict(), output: dict[str, Any] = dict(), is_error: bool = False)
dataclass
¶
Python
TriggerSpec(type: TriggerTypeEnum = MANUAL, cron: str | None = None, webhook_path: str | None = None, event_topic: str | None = None, payload_template: str = '', metadata: dict = dict())
dataclass
¶
Python
WorkerInput(instruction: str, context_messages: list[Message] = list(), metadata: dict[str, Any] = dict())
dataclass
¶
Other · DTOs¶
Bases: BaseModel
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
execution_pattern: ExecutionPatternEnum = REACT
class-attribute
instance-attribute
¶
Bases: BaseModel
Bases: BaseModel
Bases: BaseModel
Other · Enums¶
Bases: str, Enum
Bases: str, Enum
Bases: str, Enum
Bases: str, Enum
Bases: str, Enum
Bases: str, Enum
Other · Exceptions¶
Other · Protocols (ports)¶
Bases: Protocol
Contract every AI agent engine adapter must implement.
async
¶
Python
build(agent: Agent, deps: EngineDependencies) -> CompiledAgent
async
¶
Python
run(compiled: CompiledAgent, input: AgentInput) -> AgentOutput
async
¶
Python
stream(compiled: CompiledAgent, input: AgentInput) -> AsyncIterator[AgentEvent]
async
¶
Python
shutdown(compiled: CompiledAgent) -> None
Bases: Protocol
async
¶
Python
get(conversation_id: str) -> Conversation | None
async
¶
Python
list_by_agent(agent_slug: str) -> list[Conversation]
async
¶
Python
save(conversation: Conversation) -> Conversation
async
¶
Bases: Protocol
async
¶
Python
get(trace_id: str) -> ExecutionTrace | None
async
¶
Python
list_by_agent(agent_slug: str) -> list[ExecutionTrace]
async
¶
Python
save(trace: ExecutionTrace) -> ExecutionTrace
Bases: Protocol
async
¶
Python
check_input(text: str) -> GuardrailResult
async
¶
Python
check_output(text: str) -> GuardrailResult
Bases: Protocol
Surface used by the HIL runner to ask a human for go/no-go.
async
¶
Python
review(request: ApprovalRequest) -> ApprovalDecision
Bases: Protocol
Python
build_input(instruction: str, history: list[Message], target_worker_slug: str) -> WorkerInput
Python
parse_output(raw_output: str, source_worker_slug: str) -> WorkerResult
Bases: Protocol
async
¶
Python
chat(request: LLMRequest) -> LLMResponse
async
¶
Python
stream(request: LLMRequest) -> AsyncIterator[str]
Bases: Protocol
async
¶
Python
emit_trace(trace: ExecutionTrace) -> None
async
¶
async
¶
Bases: Protocol
async
¶
Python
execute(call: ToolCall) -> ToolResult
Other · Use cases¶
Python
AppendMessageUseCase(repo: IConversationRepository)
Source code in apogee_ai/application/use_cases/conversations/conversation_use_cases.py
async
¶
Python
execute(conversation_id: str, message: Message) -> Conversation
Source code in apogee_ai/application/use_cases/conversations/conversation_use_cases.py
Python
async def execute(self, conversation_id: str, message: Message) -> Conversation:
conv = await self._repo.get(conversation_id)
if conv is None:
raise KeyError(f"conversation '{conversation_id}' not found")
if conv.closed:
raise ValueError(f"conversation '{conversation_id}' is closed")
conv.append(message)
return await self._repo.save(conv)
Python
CloseConversationUseCase(repo: IConversationRepository)
Source code in apogee_ai/application/use_cases/conversations/conversation_use_cases.py
async
¶
Python
execute(conversation_id: str) -> Conversation
Source code in apogee_ai/application/use_cases/conversations/conversation_use_cases.py
Python
CreateAgentUseCase(query_repo: IAgentQueryRepository, command_repo: IAgentCommandRepository)
Source code in apogee_ai/application/use_cases/create_agent_use_case.py
async
¶
Python
execute(dto: CreateAgentDTO) -> AgentOutputDTO
Source code in apogee_ai/application/use_cases/create_agent_use_case.py
Python
DeleteAgentUseCase(query_repo: IAgentQueryRepository, command_repo: IAgentCommandRepository)
Source code in apogee_ai/application/use_cases/delete_agent_use_case.py
async
¶
Python
ExecuteAgentUseCase(query_repo: IAgentQueryRepository, engine_resolver: Callable[[Any], IAgentEngine], llm_resolver: Callable[[Any], ILLMClient], tool_resolver: Callable[[Any], Any] | None = None, guardrail_resolver: Callable[[Any], Any] | None = None, trace_repo: Any | None = None, observability_emitter: Any | None = None)
Source code in apogee_ai/application/use_cases/execute_agent_use_case.py
Python
def __init__(
self,
query_repo: IAgentQueryRepository,
engine_resolver: Callable[[Any], IAgentEngine],
llm_resolver: Callable[[Any], ILLMClient],
tool_resolver: Callable[[Any], Any] | None = None,
guardrail_resolver: Callable[[Any], Any] | None = None,
trace_repo: Any | None = None,
observability_emitter: Any | None = None,
) -> None:
self._query_repo = query_repo
self._engine_resolver = engine_resolver
self._llm_resolver = llm_resolver
self._tool_resolver = tool_resolver
self._guardrail_resolver = guardrail_resolver
self._trace_repo = trace_repo
self._emitter = observability_emitter
async
¶
Python
execute(dto: ExecuteAgentInputDTO) -> ExecuteAgentOutputDTO
Source code in apogee_ai/application/use_cases/execute_agent_use_case.py
Python
async def execute(self, dto: ExecuteAgentInputDTO) -> ExecuteAgentOutputDTO:
agent = await self._query_repo.get_by_slug(dto.slug)
if agent is None:
raise AgentNotFoundException(dto.slug)
guardrails = self._guardrail_resolver(agent) if self._guardrail_resolver else None
user_input = dto.input
if guardrails is not None:
input_check = await guardrails.apply_input(user_input)
user_input = input_check.redacted_text
engine: IAgentEngine = self._engine_resolver(agent)
llm: ILLMClient = self._llm_resolver(agent)
tool_exec = self._tool_resolver(agent) if self._tool_resolver else None
deps = EngineDependencies(llm_client=llm, tool_executor=tool_exec)
compiled = await engine.build(agent, deps)
history = [
Message(role=RoleEnum(m["role"]), content=m["content"])
for m in dto.history
]
agent_input = AgentInput(
user_input=user_input,
history=history,
metadata=dto.metadata,
)
started = time.time()
try:
output = await engine.run(compiled, agent_input)
finally:
await engine.shutdown(compiled)
duration_ms = int((time.time() - started) * 1000)
final = output.final_response
if guardrails is not None:
output_check = await guardrails.apply_output(final)
final = output_check.redacted_text
if self._trace_repo is not None:
trace = ExecutionTrace(
agent_slug=agent.slug,
user_input=dto.input,
final_response=final,
iterations=output.iterations,
finish_reason=output.finish_reason,
metadata={**output.metadata, "duration_ms": duration_ms},
)
trace.add_step(
TraceStep(
kind="agent_run",
name=agent.slug,
duration_ms=duration_ms,
input={"user_input": dto.input},
output={"final_response": final},
)
)
await self._trace_repo.save(trace)
if self._emitter is not None:
await self._emitter.emit_trace(trace)
elif self._emitter is not None:
trace = ExecutionTrace(
agent_slug=agent.slug,
user_input=dto.input,
final_response=final,
iterations=output.iterations,
finish_reason=output.finish_reason,
metadata={**output.metadata, "duration_ms": duration_ms},
)
await self._emitter.emit_trace(trace)
if self._emitter is not None:
await self._emitter.emit_metric(
"agent.iterations", float(output.iterations), {"agent": agent.slug}
)
await self._emitter.emit_metric(
"agent.duration_ms", float(duration_ms), {"agent": agent.slug}
)
return ExecuteAgentOutputDTO(
final_response=final,
iterations=output.iterations,
finish_reason=output.finish_reason,
metadata=output.metadata,
)
Python
GetAgentUseCase(query_repo: IAgentQueryRepository)
Source code in apogee_ai/application/use_cases/get_agent_use_case.py
async
¶
Python
execute(slug: str) -> AgentOutputDTO
Python
IngestKnowledgeUseCase(embedder: IEmbeddingClient, store: IVectorStore)
Source code in apogee_ai/application/use_cases/knowledge/knowledge_use_cases.py
async
¶
Python
execute(req: IngestRequest) -> int
Source code in apogee_ai/application/use_cases/knowledge/knowledge_use_cases.py
Python
async def execute(self, req: IngestRequest) -> int:
embeddings = await self._embedder.embed(req.texts)
metas = req.metadatas or [{} for _ in req.texts]
if len(metas) != len(req.texts):
raise ValueError("metadatas length must match texts length")
docs = [
Document(text=t, metadata=m, embedding=e)
for t, m, e in zip(req.texts, metas, embeddings)
]
return await self._store.upsert(docs)
Python
ListAgentsUseCase(query_repo: IAgentQueryRepository)
Source code in apogee_ai/application/use_cases/list_agents_use_case.py
async
¶
Python
execute() -> list[AgentOutputDTO]
Python
QueryKnowledgeUseCase(embedder: IEmbeddingClient, store: IVectorStore)
Source code in apogee_ai/application/use_cases/knowledge/knowledge_use_cases.py
async
¶
Python
execute(query: str, top_k: int = 5, threshold: float = 0.0) -> list[RetrievedDocument]
Source code in apogee_ai/application/use_cases/knowledge/knowledge_use_cases.py
Python
StartConversationUseCase(repo: IConversationRepository)
Source code in apogee_ai/application/use_cases/conversations/conversation_use_cases.py
async
¶
Python
execute(agent_slug: str, user_id: str | None = None, title: str = '') -> Conversation
Source code in apogee_ai/application/use_cases/conversations/conversation_use_cases.py
Python
StreamAgentUseCase(query_repo: IAgentQueryRepository, engine_resolver: Callable[[Agent], IAgentEngine], llm_resolver: Callable[[Agent], ILLMClient])
Source code in apogee_ai/application/use_cases/stream_agent_use_case.py
async
¶
Python
execute(dto: ExecuteAgentInputDTO) -> AsyncIterator[AgentEvent]
Source code in apogee_ai/application/use_cases/stream_agent_use_case.py
Python
async def execute(self, dto: ExecuteAgentInputDTO) -> AsyncIterator[AgentEvent]:
agent = await self._repo.get_by_slug(dto.slug)
if agent is None:
raise AgentNotFoundException(slug=dto.slug)
engine = self._resolve_engine(agent)
deps = EngineDependencies(llm_client=self._resolve_llm(agent))
compiled = await engine.build(agent, deps)
async def _gen() -> AsyncIterator[AgentEvent]:
try:
async for event in engine.stream(
compiled, AgentInput(user_input=dto.input)
):
yield event
finally:
await engine.shutdown(compiled)
return _gen()
Python
UpdateAgentUseCase(query_repo: IAgentQueryRepository, command_repo: IAgentCommandRepository)
Source code in apogee_ai/application/use_cases/update_agent_use_case.py
async
¶
Python
execute(slug: str, dto: UpdateAgentDTO) -> AgentOutputDTO
Source code in apogee_ai/application/use_cases/update_agent_use_case.py
Python
async def execute(self, slug: str, dto: UpdateAgentDTO) -> AgentOutputDTO:
agent = await self._query_repo.get_by_slug(slug)
if agent is None:
raise AgentNotFoundException(slug)
if dto.name is not None:
agent.name = dto.name
if dto.description is not None:
agent.description = dto.description
if dto.system_prompt is not None:
agent.prompt.system = dto.system_prompt
if dto.status is not None:
agent.status = dto.status
if dto.tags is not None:
agent.tags = list(dto.tags)
updated = await self._command_repo.update(agent)
return agent_to_output_dto(updated)