Agent Framework

@dataclass
class UnifiedAgent:
    """Orchestrator-agnostic agent configuration."""

    # Identity
    name: str                       # Unique agent name (e.g., "FraudAgent")
    description: str = ""           # Human-readable description

    # Greetings (Jinja2 templates)
    greeting: str = ""              # Initial greeting when agent takes over
    return_greeting: str = ""       # Greeting when returning to this agent

    # Handoff Configuration
    handoff: HandoffConfig          # How other agents route to this one

    # Model Settings
    model: ModelConfig              # LLM deployment, temperature, etc.

    # Voice Settings (TTS)
    voice: VoiceConfig              # Azure TTS voice name, style, rate

    # Speech Recognition (STT)
    speech: SpeechConfig            # VAD settings, languages, diarization

    # Session Settings (VoiceLive-specific)
    session: Dict[str, Any]         # Realtime API session configuration

    # Prompt
    prompt_template: str = ""       # Jinja2 prompt template (system message)

    # Tools
    tool_names: List[str]           # References to shared tool registry

    # Template Variables
    template_vars: Dict[str, Any]   # Variables for Jinja2 rendering

@dataclass
class HandoffConfig:
    trigger: str = ""           # Tool name that routes TO this agent
    is_entry_point: bool = False  # Whether this is the default starting agent

from registries.scenariostore.loader import (
    load_scenario,
    build_handoff_map_from_scenario,
    get_handoff_config,
    get_scenario_agents,
)

# Load scenario configuration
scenario = load_scenario("banking")

# Build handoff routing map
handoff_map = build_handoff_map_from_scenario("banking")
# → {"handoff_fraud_agent": "FraudAgent", ...}

# Get handoff behavior for a specific route
cfg = get_handoff_config("banking", "Concierge", "handoff_investment_advisor")
# → HandoffConfig(type="discrete", greet_on_switch=False)

# Get agents with scenario overrides applied
agents = get_scenario_agents("banking")

# registries/toolstore/fraud.py
from registries.toolstore.registry import register_tool

analyze_transactions_schema = {
    "name": "analyze_recent_transactions",
    "description": "Analyze recent transactions for suspicious patterns",
    "parameters": {
        "type": "object",
        "properties": {
            "client_id": {"type": "string"},
            "days": {"type": "integer", "default": 30}
        },
        "required": ["client_id"]
    }
}

async def analyze_recent_transactions(args: Dict[str, Any]) -> Dict[str, Any]:
    """Analyze transactions for fraud patterns."""
    client_id = args.get("client_id")
    days = args.get("days", 30)
    # ... implementation
    return {"suspicious_count": 0, "flagged_transactions": []}

# Register at module load
register_tool(
    name="analyze_recent_transactions",
    schema=analyze_transactions_schema,
    executor=analyze_recent_transactions,
    tags={"fraud", "analysis"}
)

async def handoff_fraud_agent(args: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "handoff": True,
        "target_agent": "FraudAgent",
        "message": "Let me connect you with our fraud specialist.",
        "handoff_summary": "Fraud investigation: unauthorized charge",
        "handoff_context": {
            "client_id": args.get("client_id"),
            "fraud_type": args.get("fraud_type"),
            "handoff_timestamp": datetime.utcnow().isoformat()
        }
    }

from apps.artagent.backend.agents.loader import discover_agents, build_handoff_map

# Auto-discover all agents from the agents/ directory
agents: Dict[str, UnifiedAgent] = discover_agents()
# → {"Concierge": UnifiedAgent(...), "FraudAgent": UnifiedAgent(...), ...}

# Build handoff routing map
handoff_map: Dict[str, str] = build_handoff_map(agents)
# → {"handoff_concierge": "Concierge", "handoff_fraud_agent": "FraudAgent", ...}

# In CascadeOrchestrator or LiveOrchestrator
agent = agents[current_agent_name]

# Get OpenAI-compatible tool schemas
tools = agent.get_tools()

# Render system prompt with runtime context
system_prompt = agent.render_prompt({
    "caller_name": "John",
    "session_profile": session_data,
    "customer_intelligence": intel_data
})

# Check if a tool call is a handoff
if handoff_map.get(tool_name):
    target_agent = handoff_map[tool_name]
    # Execute handoff...

from apps.artagent.backend.agents.session_manager import SessionAgentManager

# Create session manager
mgr = SessionAgentManager(
    session_id="session_123",
    base_agents=discover_agents(),
    memo_manager=memo
)

# Get agent with any session overrides applied
agent = mgr.get_agent("Concierge")

# Modify prompt at runtime
mgr.update_agent_prompt("Concierge", "You are now a Spanish-speaking assistant...")

# Modify voice
mgr.update_agent_voice("Concierge", VoiceConfig(name="es-ES-AlvaroNeural"))

# Modify available tools
mgr.update_agent_tools("Concierge", ["get_account_summary", "escalate_human"])

# Persist to Redis
await mgr.persist()

# In tools/handoffs.py
register_tool(
    "handoff_my_agent",
    handoff_my_agent_schema,
    handoff_my_agent,
    is_handoff=True
)

# Agent loading
from registries.agentstore.loader import discover_agents, build_handoff_map

# Scenario loading
from registries.scenariostore.loader import (
    load_scenario,
    build_handoff_map_from_scenario,
    get_handoff_config,
    get_scenario_agents,
)

# Tool registry
from registries.toolstore.registry import execute_tool, get_tools_for_agent