Skip to content

Quickstart — your first agent

An agent in apogee-ai is an Agent entity plus three resolvers that say, at call time, which engine, which LLM client and which tools to use. Nothing is hard-wired, which is why the same agent definition runs against the native engine in tests and against LangGraph plus a real provider in production.

This example runs offline: EchoLLMClient and InMemoryAgentRepository need no API key and no database.

quickstart.py
import asyncio

from apogee_ai import (
    Agent,
    EngineBinding,
    EngineKindEnum,
    ExecuteAgentInputDTO,
    ExecuteAgentUseCase,
    ExecutionPatternEnum,
)
from apogee_ai.infrastructure.agent_runtime.engines.native.apogee_native_engine import (
    ApogeeNativeEngine,
)
from apogee_ai.infrastructure.llm_providers.echo_llm_client import EchoLLMClient
from apogee_ai.infrastructure.persistence.in_memory_agent_repository import (
    InMemoryAgentRepository,
)


async def main() -> None:
    repository = InMemoryAgentRepository()
    await repository.create(
        Agent(
            name="Weather bot",
            slug="weather-bot",
            execution_pattern=ExecutionPatternEnum.CONVERSATIONAL,
            engine=EngineBinding(kind=EngineKindEnum.NATIVE),
        )
    )

    execute = ExecuteAgentUseCase(
        query_repo=repository,
        engine_resolver=lambda _agent: ApogeeNativeEngine(),
        llm_resolver=lambda _agent: EchoLLMClient(),
    )

    result = await execute.execute(
        ExecuteAgentInputDTO(slug="weather-bot", input="What is the weather in Lisbon?")
    )
    print("response:  ", result.final_response)
    print("iterations:", result.iterations, "finish:", result.finish_reason)


asyncio.run(main())
Output
response:   [echo] What is the weather in Lisbon?
iterations: 1 finish: stop

What each piece does

Piece Role
Agent The definition: name, slug, execution pattern, engine binding, model config
EngineBinding Which runtime executes the loop — EngineKindEnum.NATIVE or LANGGRAPH
ExecutionPatternEnum CONVERSATIONAL, REACT, and the other loop shapes
IAgentQueryRepository Where agent definitions live; InMemoryAgentRepository for tests
engine_resolver / llm_resolver Called per execution, so one agent can run on different engines and models
ExecuteAgentUseCase Loads the agent, resolves its dependencies, runs the loop

Going to production

Swap the two resolvers; the agent definition and the use case stay as they are.

container.py
from apogee_ai_providers import Provider, ProviderCredentials, ProviderFactory

execute = ExecuteAgentUseCase(
    query_repo=sql_agent_repository,
    engine_resolver=lambda agent: engine_for(agent.engine.kind),
    llm_resolver=lambda agent: ProviderFactory.build(
        Provider.ANTHROPIC,
        ProviderCredentials(api_key=settings.anthropic_key),
    ),
    tool_resolver=lambda agent: tool_registry,
    guardrail_resolver=lambda agent: guardrails,
    trace_repo=trace_repository,
    observability_emitter=otel_emitter,
)

The optional arguments are the extension points: tool_resolver supplies the callable tools, guardrail_resolver filters input and output, and trace_repo plus observability_emitter record what happened.

Streaming

StreamAgentUseCase takes the same constructor arguments and yields AgentEvents instead of returning a single DTO.

Python
async for event in stream_use_case.execute(
    ExecuteAgentInputDTO(slug="weather-bot", input="Count to five.")
):
    print(event)