API Reference¶

class AgentCapability:
    """Abstract capability labels mapped to concrete models by ``ModelRouter``.

    Use these constants in the ``@agent`` decorator to declare what kind of
    model an agent needs.  The ``ModelRouter`` translates these labels into
    provider-specific model names at runtime, allowing you to swap models
    globally without touching agent definitions.

    Attributes:
        FAST: Optimised for low latency (default model: ``gpt-5-mini``).
        SMART: Balanced quality and cost (default model: ``gpt-5.1``).
        REASONING: Advanced reasoning tasks (default model: ``gemini-3-pro``).
        VISION: Vision / multimodal support (default model: ``gpt-4o``).
        CODING: Code generation (default model: ``claude-3-5-sonnet``).

    Example:
        >>> from pico_agent import agent, AgentCapability
        >>> @agent(name="fast_bot", capability=AgentCapability.FAST)
        ... class FastBot(Protocol):
        ...     def run(self, q: str) -> str: ...
    """

    FAST = "fast"
    SMART = "smart"
    REASONING = "reasoning"
    VISION = "vision"
    CODING = "coding"

@dataclass
class AgentConfig:
    """Complete configuration for a single agent.

    Instances are created automatically by the ``@agent`` decorator and stored
    in ``LocalAgentRegistry``.  The ``AgentConfigService`` merges local, remote
    (central), and runtime overrides to produce the final effective config.

    Args:
        name: Unique agent identifier (required).
        system_prompt: System-level prompt sent to the LLM.
        user_prompt_template: Template for the user message.  Use ``{input}``
            or any key matching the method signature.
        description: Human-readable description; used as ``AgentAsTool``
            description when the agent is exposed as a tool.
        capability: ``AgentCapability`` constant that the ``ModelRouter``
            resolves to a concrete model name.
        enabled: Whether the agent is active.  Disabled agents raise
            ``AgentDisabledError``.
        agent_type: Execution strategy (``ONE_SHOT``, ``REACT``, or
            ``WORKFLOW``).
        max_iterations: Maximum ReAct loop iterations (only relevant for
            ``REACT`` agents).
        tools: List of tool names to attach to this agent.
        agents: List of child agent names that will be wrapped as
            ``AgentAsTool`` instances.
        tags: Tags used for dynamic tool lookup via ``ToolRegistry``.
        tracing_enabled: Whether ``TraceService`` records runs for this agent.
        temperature: LLM sampling temperature (0.0 -- 2.0).
        max_tokens: Maximum tokens in the LLM response, or ``None`` for the
            provider default.
        llm_profile: Named API-key / base-URL profile in ``LLMConfig``.
        workflow_config: Extra parameters for ``WORKFLOW`` agents (e.g.,
            ``{"type": "map_reduce", "splitter": "...", "reducer": "..."}``).
    """

    name: str
    system_prompt: str = ""
    user_prompt_template: str = "{input}"
    description: str = ""
    capability: str = AgentCapability.SMART
    enabled: bool = True
    agent_type: AgentType = AgentType.ONE_SHOT
    max_iterations: int = 5
    tools: List[str] = field(default_factory=list)
    agents: List[str] = field(default_factory=list)
    tags: List[str] = field(default_factory=list)
    tracing_enabled: bool = True
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    llm_profile: Optional[str] = None
    workflow_config: Dict[str, Any] = field(default_factory=dict)

class AgentType(str, Enum):
    """Execution strategy for an agent.

    Determines how the agent processes requests and interacts with tools.

    Attributes:
        ONE_SHOT: Single LLM call with no tool loop (default).
        REACT: Iterative ReAct tool loop via LangGraph, up to
            ``max_iterations`` rounds.
        WORKFLOW: Custom workflow execution (e.g., map-reduce).
    """

    ONE_SHOT = "one_shot"
    REACT = "react"
    WORKFLOW = "workflow"

@dataclass
class LLMConfig:
    """Centralised API-key and base-URL store for all LLM providers.

    ``AgentLocator`` registers a default (empty) ``LLMConfig`` singleton via
    ``@provides``.  To populate it with your credentials, use ``@configure``
    on a component method that receives ``LLMConfig`` as a parameter.  Do
    **not** register your own ``LLMConfig`` with ``@factory`` + ``@provides``
    -- that would conflict with the singleton already provided by
    ``AgentInfrastructureFactory``.

    Args:
        api_keys: Mapping of provider name (or profile name) to API key.
            Standard keys: ``"openai"``, ``"anthropic"``, ``"google"``,
            ``"azure"``, ``"deepseek"``, ``"qwen"``.
        base_urls: Mapping of provider name (or profile name) to base URL
            override.

    Example:
        >>> from pico_ioc import component, configure
        >>> from pico_agent import LLMConfig
        >>> @component
        ... class AppConfig:
        ...     @configure
        ...     def setup(self, llm: LLMConfig):
        ...         llm.api_keys["openai"] = "sk-..."
    """

    api_keys: Dict[str, str] = field(default_factory=dict)
    base_urls: Dict[str, str] = field(default_factory=dict)

class AgentConfigurationError(AgentError):
    """Raised for missing or invalid agent / provider configuration.

    Common causes include missing API keys in ``LLMConfig`` or unknown
    provider names.
    """

    pass

class AgentDisabledError(AgentError):
    """Raised when an agent is invoked but its configuration has ``enabled=False``.

    The error message follows the pattern:

        ``Agent '<name>' is disabled via configuration.``

    Args:
        agent_name: The name of the disabled agent.
    """

    def __init__(self, agent_name: str):
        super().__init__(f"Agent '{agent_name}' is disabled via configuration.")

class AgentError(Exception):
    """Base exception for all pico-agent errors."""

    pass

class AgentLifecycleError(AgentError):
    """Raised when an operation violates the agent system lifecycle.

    For example, attempting to use the system before it has reached the
    ``READY`` phase.
    """

    pass

@component(scope="singleton")
class ExperimentRegistry:
    """Singleton registry for A/B experiments on agents.

    Register experiments mapping a public name to weighted variants.  When
    ``AgentLocator`` resolves an agent name, it passes through
    ``resolve_variant()`` to select the appropriate variant.

    Example:
        >>> registry = container.get(ExperimentRegistry)
        >>> registry.register_experiment("summarizer", {
        ...     "summarizer_v1": 0.8,
        ...     "summarizer_v2": 0.2,
        ... })
    """

    def __init__(self):
        self._experiments: Dict[str, List[Tuple[str, float]]] = {}

    def register_experiment(self, public_name: str, variants: Dict[str, float]):
        """Register an A/B experiment.

        Weights are normalised so they sum to 1.0.

        Args:
            public_name: The public agent name that triggers variant
                selection.
            variants: Mapping of variant agent names to their relative
                weights.
        """
        total_weight = sum(variants.values())
        normalized_variants = []

        for name, weight in variants.items():
            normalized_variants.append((name, weight / total_weight))

        self._experiments[public_name] = normalized_variants

    def resolve_variant(self, name: str) -> str:
        """Resolve a public name to a variant using weighted random selection.

        If no experiment is registered for *name*, the name itself is
        returned unchanged.

        Args:
            name: The public agent name.

        Returns:
            The selected variant name, or *name* if no experiment exists.
        """
        if name not in self._experiments:
            return name

        variants = self._experiments[name]
        choices = [v[0] for v in variants]
        weights = [v[1] for v in variants]

        return random.choices(choices, weights=weights, k=1)[0]

class LLM(Protocol):
    """Protocol for a language-model adapter used by agent proxies.

    ``LangChainAdapter`` is the built-in implementation.
    """

    def invoke(self, messages: List[Dict[str, str]], tools: List[Any]) -> str:
        """Send messages to the LLM and return the text response.

        Args:
            messages: List of message dicts with ``"role"`` and ``"content"``
                keys.
            tools: LangChain-compatible tool instances bound to the model.

        Returns:
            The LLM's text response.
        """
        ...

    def invoke_structured(self, messages: List[Dict[str, str]], tools: List[Any], output_schema: Type[Any]) -> Any:
        """Send messages and parse the response into a structured schema.

        Args:
            messages: List of message dicts.
            tools: LangChain-compatible tool instances.
            output_schema: A ``pydantic.BaseModel`` subclass for structured
                output.

        Returns:
            An instance of *output_schema*.
        """
        ...

    def invoke_agent_loop(
        self,
        messages: List[Dict[str, str]],
        tools: List[Any],
        max_iterations: int,
        output_schema: Optional[Type[Any]] = None,
    ) -> Any:
        """Run a ReAct-style tool loop via LangGraph.

        Args:
            messages: List of message dicts.
            tools: LangChain-compatible tool instances.
            max_iterations: Maximum number of reasoning iterations.
            output_schema: Optional Pydantic model for structured final
                output.

        Returns:
            The final text response, or an instance of *output_schema* if
            provided.
        """
        ...

class CentralConfigClient(Protocol):
    """Protocol for retrieving and persisting agent configuration remotely.

    The default implementation (``NoOpCentralClient``) returns ``None`` for
    all lookups, meaning only local and runtime config is used.  Provide a
    custom implementation (e.g., backed by a database or API) to enable
    central configuration management.
    """

    def get_agent_config(self, name: str) -> Optional[Any]:
        """Fetch the remote configuration for an agent.

        Args:
            name: The agent's unique identifier.

        Returns:
            An ``AgentConfig`` if one exists remotely, otherwise ``None``.
        """
        ...

    def upsert_agent_config(self, config: Any) -> None:
        """Create or update the remote configuration for an agent.

        Args:
            config: The ``AgentConfig`` to persist.
        """
        ...

class LLMFactory(Protocol):
    """Protocol for creating ``LLM`` instances from model parameters.

    ``LangChainLLMFactory`` is the built-in implementation that supports
    OpenAI, Azure, Anthropic, Google, DeepSeek, and Qwen.
    """

    def create(
        self, model_name: str, temperature: float, max_tokens: Optional[int], llm_profile: Optional[str] = None
    ) -> LLM:
        """Create an ``LLM`` instance for the given model.

        Args:
            model_name: Model identifier, optionally prefixed with a provider
                (e.g., ``"openai:gpt-5-mini"``).
            temperature: Sampling temperature (0.0 -- 2.0).
            max_tokens: Maximum response tokens, or ``None`` for the provider
                default.
            llm_profile: Named profile in ``LLMConfig`` for API key / base
                URL selection.

        Returns:
            A configured ``LLM`` instance ready for invocation.
        """
        ...

@component(scope="singleton")
class AgentSystem:
    """Lifecycle coordinator that publishes phase transitions via ``EventBus``.

    Transitions are published as ``LifecycleEvent`` instances.  The system
    moves through: ``INITIALIZING`` -> ``READY`` -> ``RUNNING`` ->
    ``SHUTTING_DOWN`` -> ``STOPPED``.
    """

    def __init__(self):
        self._phase = LifecyclePhase.INITIALIZING
        self._event_bus: EventBus | None = None

    @property
    def phase(self) -> LifecyclePhase:
        """The current lifecycle phase."""
        return self._phase

    def _transition(self, phase: LifecyclePhase, detail: str = ""):
        self._phase = phase
        if self._event_bus:
            self._event_bus.publish_sync(LifecycleEvent(phase=phase, detail=detail))

    @configure
    def _on_ready(self, container: PicoContainer):
        if container.has(EventBus):
            self._event_bus = container.get(EventBus)
        self._transition(LifecyclePhase.READY, "Container configured")
        self._transition(LifecyclePhase.RUNNING)

    @cleanup
    def _on_shutdown(self):
        self._transition(LifecyclePhase.SHUTTING_DOWN)
        self._transition(LifecyclePhase.STOPPED)

@dataclass
class LifecycleEvent(Event):
    """Event published when the system transitions between lifecycle phases.

    Args:
        phase: The new ``LifecyclePhase``.
        detail: Optional human-readable detail string.
    """

    phase: LifecyclePhase
    detail: str = ""

class LifecyclePhase(str, Enum):
    """Phases of the pico-agent system lifecycle.

    Attributes:
        INITIALIZING: Container is being built.
        SCANNING: Agents and tools are being discovered.
        READY: Container is fully configured.
        RUNNING: System is accepting requests.
        SHUTTING_DOWN: Graceful shutdown in progress.
        STOPPED: System has stopped.
    """

    INITIALIZING = "initializing"
    SCANNING = "scanning"
    READY = "ready"
    RUNNING = "running"
    SHUTTING_DOWN = "shutting_down"
    STOPPED = "stopped"

@component(scope="singleton")
class AgentLocator:
    """Primary entry point for obtaining agent proxies.

    Resolves agent names (or Protocol classes) to ``DynamicAgentProxy``
    (for code-defined agents) or ``VirtualAgentRunner`` (for YAML-defined
    / runtime agents).  Supports A/B experiment resolution via
    ``ExperimentRegistry``.

    Args:
        container: The pico-ioc container.
        config_service: Service for resolving agent configurations.
        tool_registry: Registry for tool lookup.
        llm_factory: Factory for creating LLM instances.
        local_registry: Registry of locally discovered agents.
        model_router: Capability-to-model router.
        experiment_registry: A/B experiment variant selector.
        scheduler: Concurrency scheduler for async operations.
    """

    def __init__(
        self,
        container: PicoContainer,
        config_service: AgentConfigService,
        tool_registry: ToolRegistry,
        llm_factory: LLMFactory,
        local_registry: LocalAgentRegistry,
        model_router: ModelRouter,
        experiment_registry: ExperimentRegistry,
        scheduler: PlatformScheduler,
    ):
        self.container = container
        self.config_service = config_service
        self.tool_registry = tool_registry
        self.llm_factory = llm_factory
        self.local_registry = local_registry
        self.model_router = model_router
        self.experiment_registry = experiment_registry
        self.scheduler = scheduler

    def get_agent(self, name_or_protocol: Any) -> Optional[Any]:
        """Retrieve an agent proxy by name or Protocol class.

        Resolution order:

        1. If *name_or_protocol* is a type, look up the agent name from its
           ``AGENT_META_KEY`` metadata or from ``LocalAgentRegistry``.
        2. If a string, pass through ``ExperimentRegistry.resolve_variant()``
           for A/B testing support.
        3. If a matching Protocol exists locally, create a
           ``DynamicAgentProxy``.
        4. Otherwise, attempt to create a ``VirtualAgentRunner`` from config.

        Args:
            name_or_protocol: Either an agent name string or a Protocol class.

        Returns:
            A ``DynamicAgentProxy`` or ``VirtualAgentRunner``, or ``None`` if
            no agent could be resolved.
        """
        agent_name = ""
        protocol = None

        if isinstance(name_or_protocol, type):
            protocol = name_or_protocol
            if hasattr(protocol, AGENT_META_KEY):
                agent_name = getattr(protocol, AGENT_META_KEY).name
            else:
                for n, p in self.local_registry._protocols.items():
                    if p == protocol:
                        agent_name = n
                        break
        else:
            requested_name = str(name_or_protocol)
            agent_name = self.experiment_registry.resolve_variant(requested_name)
            protocol = self.local_registry.get_protocol(agent_name)

        if not agent_name:
            return None

        if protocol:
            return self._create_proxy(agent_name, protocol)

        try:
            config = self.config_service.get_config(agent_name)
            if config:
                return VirtualAgentRunner(
                    config=config,
                    tool_registry=self.tool_registry,
                    llm_factory=self.llm_factory,
                    model_router=self.model_router,
                    container=self.container,
                    locator=self,
                    scheduler=self.scheduler,
                )
        except ValueError:
            pass

        return None

    def _create_proxy(self, name: str, protocol: Optional[Type]) -> Any:
        return DynamicAgentProxy(
            agent_name=name,
            protocol_cls=protocol,
            config_service=self.config_service,
            tool_registry=self.tool_registry,
            llm_factory=self.llm_factory,
            model_router=self.model_router,
            container=self.container,
            locator=self,
        )

    def create_proxy(self, protocol: Type) -> Any:
        """Create a ``DynamicAgentProxy`` directly from a Protocol class.

        Convenience method that extracts the agent name from the Protocol's
        ``AGENT_META_KEY`` metadata.

        Args:
            protocol: A Protocol class decorated with ``@agent``.

        Returns:
            A ``DynamicAgentProxy`` for the given Protocol.
        """
        config = getattr(protocol, AGENT_META_KEY)
        return self._create_proxy(config.name, protocol)

@component
class AgentConfigService:
    """Merges central, local, and runtime configuration for agents.

    Configuration priority (highest wins): **central > local > runtime**.
    Central config (from ``CentralConfigClient``) takes precedence over the
    local config discovered by ``AgentScanner``.  Runtime overrides set via
    ``update_agent_config()`` are applied on top of whichever base config is
    found.

    Args:
        central_client: Remote configuration client.
        local_registry: Registry populated by ``AgentScanner``.
    """

    def __init__(self, central_client: CentralConfigClient, local_registry: LocalAgentRegistry):
        self.central_client = central_client
        self.local_registry = local_registry
        self.auto_register = True
        self._runtime_overrides: Dict[str, Dict[str, Any]] = {}

    def get_config(self, name: str) -> AgentConfig:
        """Return the effective ``AgentConfig`` for the named agent.

        Merges remote, local, and runtime sources.

        Args:
            name: Agent identifier.

        Returns:
            The merged ``AgentConfig``.

        Raises:
            ValueError: If no configuration exists for the given name.
                Message: ``"No configuration found for agent: <name>"``.
        """
        remote_config = self.central_client.get_agent_config(name)
        local_config = self.local_registry.get_config(name)

        base_config = remote_config or local_config
        runtime_data = self._runtime_overrides.get(name)

        if base_config:
            if runtime_data:
                return replace(base_config, **runtime_data)
            return base_config

        elif runtime_data:
            config_data = runtime_data.copy()
            if "name" not in config_data:
                config_data["name"] = name
            return AgentConfig(**config_data)

        raise ValueError(f"No configuration found for agent: {name}")

    def update_agent_config(self, name: str, **kwargs):
        """Apply runtime overrides to an agent's configuration.

        Overrides are merged on each ``get_config()`` call.

        Args:
            name: Agent identifier.
            **kwargs: Fields of ``AgentConfig`` to override.
        """
        if name not in self._runtime_overrides:
            self._runtime_overrides[name] = {}
        self._runtime_overrides[name].update(kwargs)

    def reset_agent_config(self, name: str):
        """Remove all runtime overrides for an agent.

        Args:
            name: Agent identifier.
        """
        if name in self._runtime_overrides:
            del self._runtime_overrides[name]