Base Agents

base

Base agent classes — DataAgent, DomainAgent, CoordinatorAgent.

DataAgent extends ADK BaseAgent for deterministic CRUD (no LLM). DomainAgent and CoordinatorAgent wrap ADK LlmAgent with scoped sub-agents and tools, preserving the Ninja Stack delegation hierarchy.

DataAgent

Bases: BaseAgent

Deterministic agent that owns a single entity.

Extends ADK BaseAgent — performs CRUD operations via its scoped tool set without LLM calls (unless reasoning_level is explicitly raised).

execute

execute(
    tool_name: str,
    trace: TraceContext | None = None,
    **kwargs: Any,
) -> Any

Execute a tool by name. Raises KeyError if not in scope.

Source code in libs/ninja-agents/src/ninja_agents/base.py

def execute(
    self,
    tool_name: str,
    trace: TraceContext | None = None,
    **kwargs: Any,
) -> Any:
    """Execute a tool by name.  Raises ``KeyError`` if not in scope."""
    tool = self._tool_map.get(tool_name)
    if tool is None:
        raise KeyError(f"Tool '{tool_name}' not in scope for agent '{self.name}'. Available: {self.tool_names}")
    span = trace.start_span(self.name) if trace else None
    try:
        return invoke_tool(tool, span=span, **kwargs)
    finally:
        if trace:
            trace.finish_span(self.name)

DomainAgent

DomainAgent(
    domain: DomainSchema,
    data_agents: list[DataAgent],
    config: AgentConfig | None = None,
)

LLM-powered agent that owns a domain (group of entities).

Wraps an ADK LlmAgent whose sub_agents are the domain's DataAgent instances. Provides convenience methods for synchronous delegation and execution that work without an LLM call (useful for testing and deterministic paths).

Source code in libs/ninja-agents/src/ninja_agents/base.py

def __init__(
    self,
    domain: DomainSchema,
    data_agents: list[DataAgent],
    config: AgentConfig | None = None,
) -> None:
    self.domain = domain
    self.config = config or domain.agent_config
    self.name = f"domain_agent_{domain.name.lower()}"
    self._data_agents: dict[str, DataAgent] = {da.entity.name: da for da in data_agents}

    model = _REASONING_MODEL.get(self.config.reasoning_level, _DEFAULT_MODEL)
    self.agent = LlmAgent(
        name=self.name,
        model=model,
        description=f"Domain agent for {domain.name}",
        instruction=(
            f"You are the {domain.name} domain agent. Delegate CRUD operations to your DataAgent sub-agents."
        ),
        tools=[],
        sub_agents=list(data_agents),
    )

delegate

delegate(
    entity_name: str,
    tool_name: str,
    trace: TraceContext | None = None,
    **kwargs: Any,
) -> Any

Delegate a tool call to a DataAgent. Raises KeyError if the entity is not in this domain.

Source code in libs/ninja-agents/src/ninja_agents/base.py

def delegate(
    self,
    entity_name: str,
    tool_name: str,
    trace: TraceContext | None = None,
    **kwargs: Any,
) -> Any:
    """Delegate a tool call to a DataAgent.  Raises ``KeyError`` if
    the entity is not in this domain."""
    da = self._data_agents.get(entity_name)
    if da is None:
        raise KeyError(f"Entity '{entity_name}' not in domain '{self.domain.name}'. Available: {self.entity_names}")
    if trace:
        trace.start_span(self.name)
    try:
        return da.execute(tool_name, trace=trace, **kwargs)
    finally:
        if trace:
            trace.finish_span(self.name)

execute

execute(
    request: str, trace: TraceContext | None = None
) -> dict[str, Any]

Process a domain-level request (stub — full impl uses LLM).

Source code in libs/ninja-agents/src/ninja_agents/base.py

def execute(self, request: str, trace: TraceContext | None = None) -> dict[str, Any]:
    """Process a domain-level request (stub — full impl uses LLM)."""
    if trace:
        trace.start_span(self.name)
    try:
        return {
            "agent": self.name,
            "domain": self.domain.name,
            "request": request,
            "available_entities": self.entity_names,
            "uses_llm": self.uses_llm,
        }
    finally:
        if trace:
            trace.finish_span(self.name)

CoordinatorAgent

CoordinatorAgent(
    domain_agents: list[DomainAgent],
    config: AgentConfig | None = None,
)

Top-level routing agent that delegates to DomainAgents.

Wraps an ADK LlmAgent whose sub_agents are the underlying DomainAgent.agent instances. Uses LLM for intent classification and result synthesis.

Source code in libs/ninja-agents/src/ninja_agents/base.py

def __init__(
    self,
    domain_agents: list[DomainAgent],
    config: AgentConfig | None = None,
) -> None:
    self.config = config or AgentConfig(reasoning_level=ReasoningLevel.HIGH)
    self.name = "coordinator"
    self._domain_agents: dict[str, DomainAgent] = {da.domain.name: da for da in domain_agents}

    model = _REASONING_MODEL.get(self.config.reasoning_level, _DEFAULT_MODEL)
    self.agent = LlmAgent(
        name=self.name,
        model=model,
        description="Coordinator that routes requests across domains",
        instruction=(
            "You are the top-level coordinator. Classify the user's intent "
            "and delegate to the appropriate domain agent."
        ),
        tools=[],
        sub_agents=[da.agent for da in domain_agents],
    )

route

route(
    request: str,
    target_domains: list[str],
    trace: TraceContext | None = None,
) -> dict[str, Any]

Route a request to specific domains and collect results.

For parallel execution use Orchestrator.fan_out() instead.

Source code in libs/ninja-agents/src/ninja_agents/base.py

def route(
    self,
    request: str,
    target_domains: list[str],
    trace: TraceContext | None = None,
) -> dict[str, Any]:
    """Route a request to specific domains and collect results.

    For parallel execution use ``Orchestrator.fan_out()`` instead.
    """
    results: dict[str, Any] = {}
    for domain_name in target_domains:
        da = self._domain_agents.get(domain_name)
        if da is None:
            results[domain_name] = {"error": f"Unknown domain: {domain_name}"}
            continue
        results[domain_name] = da.execute(request, trace=trace)
    return results

create_domain_agent

create_domain_agent(
    domain: DomainSchema, data_agents: list[DataAgent]
) -> LlmAgent

Factory: create a bare ADK LlmAgent for a domain.

Source code in libs/ninja-agents/src/ninja_agents/base.py

def create_domain_agent(
    domain: DomainSchema,
    data_agents: list[DataAgent],
) -> LlmAgent:
    """Factory: create a bare ADK LlmAgent for a domain."""
    model = _REASONING_MODEL.get(domain.agent_config.reasoning_level, _DEFAULT_MODEL)
    return LlmAgent(
        name=f"domain_agent_{domain.name.lower()}",
        model=model,
        description=f"Domain agent for {domain.name} — cross-entity reasoning",
        instruction=(f"You are the {domain.name} domain agent. Delegate CRUD operations to your DataAgent sub-agents."),
        tools=[],
        sub_agents=list(data_agents),
    )

create_coordinator_agent

create_coordinator_agent(
    domain_agents: list[DomainAgent],
) -> LlmAgent

Factory: create a bare ADK LlmAgent coordinator.

Source code in libs/ninja-agents/src/ninja_agents/base.py

def create_coordinator_agent(
    domain_agents: list[DomainAgent],
) -> LlmAgent:
    """Factory: create a bare ADK LlmAgent coordinator."""
    return LlmAgent(
        name="coordinator",
        model=_DEFAULT_MODEL,
        description="Coordinator that routes requests across domains",
        instruction=("Classify the user's intent and delegate to the appropriate domain agent."),
        tools=[],
        sub_agents=[da.agent for da in domain_agents],
    )