"""FastAPI dependency injection — provides shared singletons to route handlers."""
from typing import Annotated
from fastapi import Depends
from navi.config import settings
from navi.context_providers._loader import ContextProviderRegistry
from navi.core import (
Agent,
BackendRegistry,
PgSessionStore,
ProfileRegistry,
SessionStore,
SqliteSessionStore,
ToolRegistry,
build_default_registries,
)
from navi.memory import MemoryStore
from navi.memory.sqlite_store import SqliteMemoryStore
from navi.workers import Worker, build_default_workers
def _make_session_store() -> SessionStore:
if settings.database_url:
return PgSessionStore(settings.database_url)
return SqliteSessionStore(settings.db_path)
def _make_memory_store() -> MemoryStore:
if settings.database_url:
return MemoryStore(settings.database_url)
return SqliteMemoryStore(settings.db_path)
_memory_store = _make_memory_store()
_registries: tuple[ToolRegistry, ProfileRegistry, BackendRegistry, ContextProviderRegistry] | None = None
def get_memory_store() -> MemoryStore:
return _memory_store
def get_registries() -> tuple[ToolRegistry, ProfileRegistry, BackendRegistry, ContextProviderRegistry]:
global _registries
if _registries is None:
_registries = build_default_registries(
memory_store=_memory_store,
session_store=_session_store,
)
return _registries
def get_tool_registry() -> ToolRegistry:
return get_registries()[0]
def get_profile_registry() -> ProfileRegistry:
return get_registries()[1]
def get_backend_registry() -> BackendRegistry:
return get_registries()[2]
def get_cp_registry() -> ContextProviderRegistry:
return get_registries()[3]
_session_store = _make_session_store()
_workers: list[Worker] = build_default_workers()
def get_session_store() -> SessionStore:
return _session_store
def get_workers() -> list[Worker]:
return _workers
def get_agent(
session_store: Annotated[SessionStore, Depends(get_session_store)],
profile_registry: Annotated[ProfileRegistry, Depends(get_profile_registry)],
tool_registry: Annotated[ToolRegistry, Depends(get_tool_registry)],
backend_registry: Annotated[BackendRegistry, Depends(get_backend_registry)],
cp_registry: Annotated[ContextProviderRegistry, Depends(get_cp_registry)],
) -> Agent:
return Agent(session_store, profile_registry, tool_registry, backend_registry,
workers=get_workers(), memory_store=_memory_store, cp_registry=cp_registry)