scripts-agent-invocation

#!/usr/bin/env python3 """

title: Agent Invocation with Context Injection component_type: script version: 1.0.0 status: active summary: Unified agent invocation API with automatic context injection (ADR-154) keywords: [agent, invocation, context, injection, adr-154, memory, dispatcher] track: J task_id: J.4.9.5 created: 2026-02-03

Agent Invocation Module - J.4.9.5 Implementation

Provides a unified API for agent invocation that automatically:

1. Classifies task intent from user message
2. Injects relevant context via AgentContextInjector
3. Formats the agent prompt with context
4. Returns structured invocation for Claude Code Task tool

This module integrates with:

• AgentContextInjector (J.4.9.1)
• CurriculumAgentDispatcher
• Claude Code Task protocol

Usage: from scripts.core.agent_invocation import invoke_agent, AgentInvocation

# Simple invocation with context
result = invoke_agent(
    agent_type="security-specialist",
    task="Review the authentication code for vulnerabilities",
    session_id="session-123"
)

# Use in Claude Code
print(result.task_call)  # Task(...) ready for execution

# Access injected context
print(result.context_section)

"""

import json import logging import uuid from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional

Setup logging

logger = logging.getLogger(name)

Add paths for imports

import sys _script_dir = Path(file).resolve().parent _coditect_root = _script_dir.parent.parent if str(_coditect_root) not in sys.path: sys.path.insert(0, str(_coditect_root))

Import AgentContextInjector

try: from scripts.context_graph.agent_context_injector import ( AgentContextInjector, ContextInjection, IntentClassification, classify_intent, ) INJECTOR_AVAILABLE = True except ImportError: INJECTOR_AVAILABLE = False logger.warning("AgentContextInjector not available - context injection disabled")

Import WorkflowStateManager for workflow context

try: from scripts.context_graph.workflow_state_manager import WorkflowStateManager WORKFLOW_AVAILABLE = True except ImportError: WORKFLOW_AVAILABLE = False logger.warning("WorkflowStateManager not available - workflow features disabled")

=============================================================================

Data Classes

=============================================================================

@dataclass class AgentInvocation: """Result of an agent invocation with context injection."""

agent_type: str
task: str
session_id: str

# Generated content
task_call: str                    # Claude Code Task() call
prompt_with_context: str          # Full prompt including injected context
context_section: str              # Just the context section

# Context metadata
intent: Optional[Dict[str, Any]] = None
workflow_id: Optional[str] = None
context_tokens: int = 0
templates_used: List[str] = field(default_factory=list)

# Timing
created_at: str = field(default_factory=lambda: datetime.now().isoformat())

def to_dict(self) -> Dict[str, Any]:
    """Convert to dictionary for JSON serialization."""
    return {
        "agent_type": self.agent_type,
        "task": self.task,
        "session_id": self.session_id,
        "task_call": self.task_call,
        "context_tokens": self.context_tokens,
        "templates_used": self.templates_used,
        "intent": self.intent,
        "workflow_id": self.workflow_id,
        "created_at": self.created_at,
    }

=============================================================================

Agent Type Mapping

=============================================================================

Map agent names to Claude Code subagent types

AGENT_TO_SUBAGENT: Dict[str, str] = { # Development agents "senior-architect": "general-purpose", "code-reviewer": "general-purpose", "database-architect": "general-purpose", "frontend-react-typescript-expert": "general-purpose", "backend-architect": "general-purpose",

# DevOps agents
"devops-engineer": "general-purpose",
"cloud-architect": "general-purpose",
"k8s-statefulset-specialist": "general-purpose",

# Security agents
"security-specialist": "general-purpose",
"penetration-testing-agent": "general-purpose",
"compliance-auditor": "general-purpose",

# Documentation agents
"codi-documentation-writer": "general-purpose",
"software-design-document-specialist": "general-purpose",
"qa-reviewer": "general-purpose",

# Testing agents
"testing-specialist": "general-purpose",
"qa-specialist": "general-purpose",

# Special agents
"orchestrator": "general-purpose",
"research-agent": "Explore",

# MoE agents (J.25.5.3)
"moe-content-classifier": "general-purpose",
"moe-judge": "general-purpose",
"moe-evaluator": "general-purpose",
"moe-coordinator": "general-purpose",

}

Agent personas for prompt augmentation

AGENT_PERSONAS: Dict[str, str] = { "senior-architect": "You are an experienced software architect specializing in system design, ADRs, and architecture patterns.", "security-specialist": "You are a security expert specializing in vulnerability assessment, OWASP Top 10, and secure coding practices.", "code-reviewer": "You are a thorough code reviewer focused on quality, maintainability, and best practices.", "devops-engineer": "You are a DevOps engineer specializing in CI/CD, infrastructure as code, and deployment automation.", "testing-specialist": "You are a QA specialist focused on test strategy, coverage, and test automation.", "codi-documentation-writer": "You are a technical writer specializing in clear, comprehensive documentation.", "database-architect": "You are a database architect specializing in schema design, optimization, and data modeling.", "orchestrator": "You are a project coordinator managing multi-agent workflows and task orchestration.",

# MoE agents (J.25.5.3)
"moe-content-classifier": "You are a document classification expert using Mixture of Experts analysis to determine document type, quality, and metadata with high confidence.",
"moe-judge": "You are a quality evaluator using multi-perspective assessment to verify correctness, completeness, and compliance of outputs.",
"moe-evaluator": "You are a technical evaluator using structured rubrics and evidence-based analysis to score and improve artifacts.",
"moe-coordinator": "You are an MoE orchestration coordinator managing multi-expert evaluation workflows and consensus building.",

}

=============================================================================

Core Functions

=============================================================================

def invoke_agent( agent_type: str, task: str, session_id: Optional[str] = None, workflow_id: Optional[str] = None, context_budget: int = 4000, include_context: bool = True, include_persona: bool = True, ) -> AgentInvocation: """ Invoke an agent with automatic context injection.

This is the main entry point for agent invocation. It:
1. Classifies the task intent
2. Injects relevant context (decisions, patterns, errors, etc.)
3. Builds a formatted prompt with context
4. Returns a Task() call ready for Claude Code

Args:
    agent_type: Type of agent (e.g., "security-specialist")
    task: Task description / user message
    session_id: Session ID (auto-generated if None)
    workflow_id: Workflow ID for multi-turn context
    context_budget: Token budget for context injection
    include_context: Whether to inject context (default True)
    include_persona: Whether to include agent persona (default True)

Returns:
    AgentInvocation with task_call ready for execution

Example:
    >>> result = invoke_agent("security-specialist", "Review auth code")
    >>> print(result.task_call)
    Task(
        subagent_type="general-purpose",
        description="security-specialist: Review auth code",
        prompt=\"\"\"...\"\"\"
    )
"""
# Generate session ID if not provided
if session_id is None:
    session_id = f"session-{uuid.uuid4().hex[:8]}"

# Initialize context components
context_section = ""
intent_data = None
templates_used = []
context_tokens = 0

# Step 1: Inject context if enabled and available
if include_context and INJECTOR_AVAILABLE:
    try:
        injector = AgentContextInjector(default_budget=context_budget)
        injection = injector.inject_context(
            agent_type=agent_type,
            user_message=task,
            session_id=session_id,
            workflow_id=workflow_id,
            budget_tokens=context_budget,
        )
        context_section = injection.prompt_section
        intent_data = injection.metadata.get("intent")
        templates_used = injection.templates_used
        context_tokens = injection.metadata.get("token_estimate", 0)
    except Exception as e:
        logger.error(f"Context injection failed: {e}")
        # Continue without context

# Step 2: Build agent persona
persona_section = ""
if include_persona:
    persona = AGENT_PERSONAS.get(agent_type, f"You are a {agent_type} agent.")
    persona_section = f"## Agent Role\n\n{persona}\n\n"

# Step 3: Build workflow context if available
workflow_section = ""
if workflow_id and WORKFLOW_AVAILABLE:
    try:
        manager = WorkflowStateManager()
        summary = manager.get_resume_summary(workflow_id)
        if summary:
            workflow_section = f"## Workflow Context\n\n{summary}\n\n"
    except Exception as e:
        logger.error(f"Workflow context failed: {e}")

# Step 4: Compose full prompt
prompt_parts = []

if persona_section:
    prompt_parts.append(persona_section)

if context_section:
    prompt_parts.append(context_section)

if workflow_section:
    prompt_parts.append(workflow_section)

prompt_parts.append(f"## Task\n\n{task}")

prompt_parts.append("""

Instructions

1. Analyze the task and any provided context

2. Execute the task according to your agent role

3. Report back with:

   • What was completed
   • What remains to be done (if any)
   • Key findings or recommendations
   • Any blockers or issues encountered """)

   full_prompt = "\n".join(prompt_parts)

   Step 5: Generate Task() call

   subagent_type = AGENT_TO_SUBAGENT.get(agent_type, "general-purpose") description = f"{agent_type}: {task[:50]}..." if len(task) > 50 else f"{agent_type}: {task}"

   Escape quotes in prompt for Task() call

   escaped_prompt = full_prompt.replace('"""', '\"\"\"')

   task_call = f'''Task( subagent_type="{subagent_type}", description="{description}", prompt="""{escaped_prompt}""" )'''

   return AgentInvocation( agent_type=agent_type, task=task, session_id=session_id, task_call=task_call, prompt_with_context=full_prompt, context_section=context_section, intent=intent_data, workflow_id=workflow_id, context_tokens=context_tokens, templates_used=templates_used, )

def invoke_with_dispatcher( task_description: str, requirements: Optional[Dict[str, Any]] = None, session_id: Optional[str] = None, ) -> AgentInvocation: """ Use CurriculumAgentDispatcher to select and invoke the best agent.

This function:
1. Analyzes the task with the dispatcher
2. Gets agent recommendations
3. Invokes the recommended agent with context

Args:
    task_description: Description of the task
    requirements: Optional requirements dict for dispatcher
    session_id: Session ID

Returns:
    AgentInvocation with recommended agent and context
"""
# Import dispatcher
try:
    from scripts.core.agent_dispatcher import CurriculumAgentDispatcher
except ImportError:
    logger.error("CurriculumAgentDispatcher not available")
    # Fallback to direct invocation
    return invoke_agent(
        agent_type="orchestrator",
        task=task_description,
        session_id=session_id,
    )

# Analyze and recommend
dispatcher = CurriculumAgentDispatcher()
requirements = requirements or {}

try:
    task_req = dispatcher.analyze_workflow(task_description, requirements)
    recommendation = dispatcher.recommend_agents(task_req)
    primary_agent = recommendation.primary_agent
except Exception as e:
    logger.error(f"Dispatcher analysis failed: {e}")
    primary_agent = "orchestrator"

# Invoke with context
return invoke_agent(
    agent_type=primary_agent,
    task=task_description,
    session_id=session_id,
)

def get_agent_prompt( agent_type: str, task: str, session_id: Optional[str] = None, context_budget: int = 4000, ) -> str: """ Get just the prompt with context, without Task() wrapper.

Useful when you need the prompt content for other purposes.

Args:
    agent_type: Type of agent
    task: Task description
    session_id: Session ID
    context_budget: Token budget for context

Returns:
    Formatted prompt string with context
"""
invocation = invoke_agent(
    agent_type=agent_type,
    task=task,
    session_id=session_id,
    context_budget=context_budget,
)
return invocation.prompt_with_context

=============================================================================

CLI Interface

=============================================================================

def main(): """CLI interface for agent invocation.""" import argparse

parser = argparse.ArgumentParser(
    description="Invoke agents with automatic context injection"
)
subparsers = parser.add_subparsers(dest="command", help="Commands")

# invoke command
invoke_parser = subparsers.add_parser("invoke", help="Invoke an agent")
invoke_parser.add_argument("agent_type", help="Type of agent")
invoke_parser.add_argument("task", help="Task description")
invoke_parser.add_argument("--session", help="Session ID")
invoke_parser.add_argument("--workflow", help="Workflow ID")
invoke_parser.add_argument("--budget", type=int, default=4000, help="Token budget")
invoke_parser.add_argument("--no-context", action="store_true", help="Skip context injection")
invoke_parser.add_argument("--json", action="store_true", help="Output as JSON")
invoke_parser.add_argument("--prompt-only", action="store_true", help="Output only prompt")

# dispatch command
dispatch_parser = subparsers.add_parser("dispatch", help="Auto-select and invoke agent")
dispatch_parser.add_argument("task", help="Task description")
dispatch_parser.add_argument("--session", help="Session ID")
dispatch_parser.add_argument("--json", action="store_true", help="Output as JSON")

# list-agents command
list_parser = subparsers.add_parser("list-agents", help="List available agents")
list_parser.add_argument("--json", action="store_true", help="Output as JSON")

args = parser.parse_args()

if args.command == "invoke":
    result = invoke_agent(
        agent_type=args.agent_type,
        task=args.task,
        session_id=args.session,
        workflow_id=args.workflow,
        context_budget=args.budget,
        include_context=not args.no_context,
    )

    if args.json:
        print(json.dumps(result.to_dict(), indent=2))
    elif args.prompt_only:
        print(result.prompt_with_context)
    else:
        print("=" * 60)
        print(f"Agent: {result.agent_type}")
        print(f"Session: {result.session_id}")
        print(f"Context Tokens: {result.context_tokens}")
        print(f"Templates: {', '.join(result.templates_used) or 'none'}")
        print("=" * 60)
        print("\nTask Call:\n")
        print(result.task_call)

elif args.command == "dispatch":
    result = invoke_with_dispatcher(
        task_description=args.task,
        session_id=args.session,
    )

    if args.json:
        print(json.dumps(result.to_dict(), indent=2))
    else:
        print(f"Recommended Agent: {result.agent_type}")
        print(f"Context Tokens: {result.context_tokens}")
        print("\nTask Call:\n")
        print(result.task_call)

elif args.command == "list-agents":
    agents = list(AGENT_TO_SUBAGENT.keys())
    if args.json:
        print(json.dumps({"agents": agents}, indent=2))
    else:
        print("Available Agents:")
        for agent in sorted(agents):
            persona = AGENT_PERSONAS.get(agent, "No persona defined")
            print(f"  - {agent}")
            print(f"    {persona[:60]}...")

else:
    parser.print_help()

if name == "main": main()