Newer
Older
navi-1 / navi / core / agent / core.py
"""The assembled `Agent` class — behaviour lives in the mixins."""

from typing import TYPE_CHECKING

import structlog

from ..compressor import ContextCompressor
from ..context_builder import ContextBuilder
from ..planning import PlanningEngine
from ..registry import BackendRegistry, ProfileRegistry, ToolRegistry
from ..session import SessionStore
from ..subagent_runner import SubAgentRunner
from ..tool_executor import ToolExecutor

if TYPE_CHECKING:
    from navi.context_providers._loader import ContextProviderRegistry
    from navi.memory.store import MemoryStore
    from navi.workers.base import Worker


log = structlog.get_logger()

from .compression import AgentCompressionMixin
from .context import AgentContextMixin
from .runtime import AgentRuntimeMixin
from .streaming import AgentStreamMixin
from .tools import AgentToolsMixin


class Agent(
    AgentRuntimeMixin,
    AgentStreamMixin,
    AgentToolsMixin,
    AgentCompressionMixin,
    AgentContextMixin,
):
    def __init__(
        self,
        session_store: "SessionStore | None",
        profile_registry: ProfileRegistry,
        tool_registry: ToolRegistry,
        backend_registry: BackendRegistry,
        workers: list["Worker"] | None = None,
        memory_store: "MemoryStore | None" = None,
        memory_search=None,
        cp_registry: "ContextProviderRegistry | None" = None,
        mcp_manager=None,
    ) -> None:
        self._sessions = session_store
        self._profiles = profile_registry
        self._tools = tool_registry
        self._backends = backend_registry
        self._workers: list["Worker"] = workers or []
        self._memory = memory_store
        self._memory_search = memory_search
        self._cp_registry = cp_registry
        self._mcp_manager = mcp_manager
        self._ctx_builder = ContextBuilder(
            profile_registry=profile_registry,
            memory_store=memory_store,
            memory_search=memory_search,
            cp_registry=cp_registry,
            mcp_manager=mcp_manager,
        )
        self._tool_executor = ToolExecutor(tool_registry)
        self._planning = PlanningEngine(self._ctx_builder)
        self._compressor = ContextCompressor()
        self._subagent = SubAgentRunner(
            profile_registry=profile_registry,
            tool_registry=tool_registry,
            backend_registry=backend_registry,
            ctx_builder=self._ctx_builder,
            compressor=self._compressor,
            planning=self._planning,
            tool_executor=self._tool_executor,
            session_store=session_store,
            mcp_manager=mcp_manager,
        )