Agents Reference

graph TB
    subgraph "Pipeline Agents (src/agents/)"
        AA[AnalyzerAgent] --> RA[RetrieverAgent]
        RA --> EA[EnrichmentAgent]
        EA --> GA[GeneratorAgent]
        GA --> IA[InterpreterAgent]
        RA -.-> CCA[CallChainAnalyzer]
        RA -.-> CFG[ControlFlowGenerator]
        GA -.-> PB[PromptBuilder]
        GA -.-> QV[QueryVariantGenerator]
        IA -.-> AG[AnswerGenerator]
        IA -.-> RP[ResultParser]
    end
    subgraph "ACP Protocol (src/acp/)"
        IDE[IDE Client] -->|JSON-RPC| ACPA[CodeGraphACPAgent]
        ACPA --> SM[SessionManager]
        ACPA --> TM[ThreadManager]
        ACPA --> TO[TurnOrchestrator]
        ACPA --> DE[DiagnosticsEngine]
        ACPA --> HP[HoverProvider]
        ACPA -->|adapters| WA[WorkflowAdapter]
        WA --> MSC[MultiScenarioCopilot]
    end
    subgraph "Domain Agents"
        SS[SecurityScanner]
        PP[PerformanceProfiler]
        DA[DependencyAnalyzer]
    end

from src.agents import AnalyzerAgent

analyzer = AnalyzerAgent(
    llm=None,                              # Optional LLM for advanced analysis
    cpg_config: Optional[CPGConfig] = None # Domain-specific config
)

# Analyze a question — returns intent, domain, keywords, confidence
analysis = analyzer.analyze("What methods handle transactions?")
# Returns: {'intent': 'find_methods', 'domain': 'transaction-manager',
#           'keywords': ['transaction'], 'query_type': 'semantic', 'confidence': 0.85}

# Classify query mode (semantic vs structural)
mode, confidence = analyzer.classify_query_mode("Find all callers of foo")
# Returns: ('structural', 0.9)

# LLM-enhanced analysis (requires llm parameter to be set)
analysis = analyzer.analyze_with_llm("Explain the transaction commit flow")
# Returns: enriched analysis dict with LLM-derived intent and domain

# Batch analysis — process multiple questions at once
results = analyzer.batch_analyze(["Q1...", "Q2...", "Q3..."])
# Returns: list of analysis dicts

# Get ChromaDB filter for domain
domain_filter = analyzer.get_domain_filter("transaction-manager")
# Returns: dict suitable for ChromaDB where clause

from src.agents import RetrieverAgent

retriever = RetrieverAgent(
    vector_store=vector_store,           # VectorStoreReal instance
    analyzer_agent=analyzer,             # AnalyzerAgent instance
    cpg_service=None,                    # Optional CPGQueryService for graph queries
    cache_size: int = 128,               # Max cached entries
    cache_ttl_seconds: Optional[float] = None,  # Cache TTL (None = no expiration)
    enable_cache_metrics: bool = True,   # Collect cache metrics
    enable_hybrid: bool = True           # Enable hybrid retrieval
)

# Basic vector retrieval
result = retriever.retrieve(question, analysis, top_k_qa=3, top_k_sql=5)

# Hybrid retrieval (vector + graph with RRF)
result = retriever.retrieve_hybrid(
    question="Find memory allocation patterns",
    analysis=analysis,
    mode="hybrid",           # hybrid | vector_only | graph_only
    query_type="structural", # Affects weight distribution
    top_k=10,
    use_ranker=True
)

# With reranking
result = retriever.retrieve_with_reranking(question, analysis,
    top_k_qa=5, top_k_sql=10, final_k_qa=3, final_k_sql=5)

# With enrichment-aware reranking
result = retriever.retrieve_with_enrichment(question, analysis, enrichment_hints)

# Keyword-based retrieval
result = retriever.retrieve_by_keywords(["lock", "acquire"], top_k=5)

# Cache management
stats = retriever.get_stats()
metrics = retriever.get_cache_metrics()
cleared = retriever.invalidate_cache(pattern="lock*")
warmed = retriever.warm_cache(["Q1", "Q2"])

{
    'results': [...],             # Raw retrieval results
    'ranked_results': [...],      # Ranked results with scores
    'retrieval_stats': {
        'total_retrieved': 15,
        'source_distribution': {'vector': 8, 'graph': 7},
        'cache_hit': False,
        'retrieval_time_ms': 42.5
    }
}

from src.agents import EnrichmentAgent

enrichment = EnrichmentAgent(
    enable_fallback: bool = True  # Enable Phase 4 fallback strategies
)

# Get enrichment hints for a question
hints = enrichment.get_enrichment_hints(
    question="How does the lock manager work?",
    analysis={'intent': 'explain-concept', 'domain': 'locking', 'keywords': ['lock']}
)
# Returns dict with: function_purposes, data_structures, concurrency_patterns,
#                     memory_management, error_handling, security_attributes, etc.

from src.agents import GeneratorAgent

generator = GeneratorAgent(
    sql_generator: SQLQueryGenerator,    # SQL generation backend
    enable_feedback: bool = True,        # Enable tag feedback tracking
    use_semantic: bool = False           # Use semantic mode
)

# Generate SQL query
query, is_valid, error = generator.generate(question, context)
# Returns: (query_string, validation_bool, error_message_or_none)

# With retries on failure
query, is_valid, error, attempts = generator.generate_with_retries(
    question, context, max_retries=2
)
# Returns: (query_string, validation_bool, error_message_or_none, attempt_count)

# Batch generation for multiple questions
results = generator.generate_batch(questions, contexts)
# Returns: list of (query, is_valid, error) tuples

# Query Funnel — generate multiple query variants
variants = generator.generate_query_variants(question, context, num_variants=3)
# Returns: [{'query': '...', 'strategy': 'precise'}, ...]

# Explain a generated query in natural language
explanation = generator.explain_query(query, context)
# Returns: string explanation of what the query does

from src.agents.interpreter_agent import InterpreterAgent

interpreter = InterpreterAgent(
    llm_interface=None,                  # Optional LLM (falls back to templates)
    use_semantic: bool = False,          # Semantic mode (extract comments)
    cpg_config: Optional[CPGConfig] = None  # Domain-specific config
)

# Interpret query results into a natural language answer
answer = interpreter.interpret(
    question="What methods call LWLockAcquire?",
    results=[...],         # Raw query results
    query="SELECT...",     # The executed query
    enrichment_hints=None  # Optional enrichment context
)
# Returns: {
#     'answer': 'LWLockAcquire is called by 15 methods including...',
#     'confidence': 0.85,
#     'sources': [{'method': 'heap_insert', 'file': 'heapam.c', 'line': 234}, ...]
# }

from src.agents.call_chain_analyzer import CallChainAnalyzer

analyzer = CallChainAnalyzer()  # No parameters

result = analyzer.analyze(
    entry_point_result={"method": "main", ...},
    keyword_methods_result=[{"method": "foo", ...}, ...],
    call_graph_result=[{"caller": "a", "callee": "b"}, ...],
    question_keywords=["transaction", "commit"]
)

{
    'entry_point': 'main',                    # Detected entry point function
    'call_graph': {                           # Adjacency list representation
        'main': ['init', 'process'],
        'process': ['validate', 'commit'],
        ...
    },
    'call_chains': [                          # Paths through the call graph
        ['main', 'process', 'commit'],
        ['main', 'process', 'validate'],
        ...
    ],
    'key_functions': ['commit', 'validate'],  # Functions matching keywords
    'metadata': {
        'total_functions': 12,
        'max_chain_depth': 5,
        'keyword_matches': 3
    }
}

from src.agents.control_flow_generator import ControlFlowGenerator

generator = ControlFlowGenerator(llm=None)

from src.agents.logic_synthesizer import LogicSynthesizer

synthesizer = LogicSynthesizer(
    llm: Optional[Any] = None,   # LLM for generating explanations
    language: str = "en"         # Language for prompts
)

result = synthesizer.synthesize(
    question="How does the shutdown sequence work?",
    call_chain_analysis=call_chain_result,  # From CallChainAnalyzer.analyze()
    max_tokens=1000
)
# Returns: {'explanation': '...', 'metadata': {...}}

from src.agents.logic_synthesizer import format_call_chain_for_prompt

formatted = format_call_chain_for_prompt(analysis_result)
# Returns dict with: entry_point, call_graph, call_chains, key_functions

from src.agents.adaptive_refiner import AdaptiveQueryRefiner

refiner = AdaptiveQueryRefiner(
    persistence_path: Optional[str] = None  # Path to persist patterns
)

# Record query outcome for learning
refiner.record_query_outcome(
    question="...", question_type="find_callers", query="SELECT...",
    success=True, result_count=15, execution_time=0.5
)

# Suggest refinements for a failed query
suggestions = refiner.suggest_refinements(
    question="...", question_type="find_callers",
    failed_query="SELECT...", max_suggestions=3
)
# Returns: list of suggested query strings

# Get best pattern for a question type
pattern = refiner.get_best_pattern_for_type("find_callers")
# Returns: QueryPattern or None

# Statistics
stats = refiner.get_statistics()
# Returns: {'total_patterns': 25, 'avg_success_rate': 0.78, ...}

# Persistence
refiner.save_patterns()
refiner.load_patterns()

from src.agents.answer_generator import AnswerGenerator

generator = AnswerGenerator(
    llm_interface=None,                    # Optional LLM
    code_analyst_title: str = "code analyst",  # Domain-specific title
    language: str = "en"                   # Prompt language
)

# Group functions by enrichment tags
grouped = generator.group_by_tags(functions, enrichment_hints)
# Returns: dict mapping tag categories to lists of functions

# Generate answer from grouped results
answer = generator.generate(question, query, grouped, enrichment_hints,
                            used_fallback=False, fallback_query=None)
# Returns: string answer

from src.agents.answer_generator import create_answer_generator

generator = create_answer_generator(
    llm_interface=llm,
    code_analyst_title="code analyst",
    language="en"
)
# Returns: AnswerGenerator instance

from src.agents.result_reranker import ResultReranker, create_qa_reranker

# Using factory functions
reranker = create_qa_reranker()
reranker = create_sql_reranker()
reranker = create_enrichment_reranker()

# Or direct construction with a strategy
reranker = ResultReranker(strategy=QARerankerStrategy())

# Rerank items
ranked = reranker.rerank(items, analysis, top_k=5)
# Returns: list of items sorted by relevance score, trimmed to top_k

RerankerStrategy (ABC)
  +-- QARerankerStrategy      # Q&A pair reranking
  +-- SQLRerankerStrategy     # SQL example reranking
  +-- EnrichmentRerankerStrategy  # Enrichment-aware reranking

from src.agents.result_reranker import ScoringBoost

boost = ScoringBoost(
    name="domain_match",       # Human-readable boost name
    multiplier=1.5,            # Score multiplier when condition is true
    condition=lambda item, analysis: item.get("domain") == analysis.get("domain")
)

from src.agents.confidence_calculator import (
    ConfidenceCalculator, create_result_calculator,
    create_analysis_calculator, create_pattern_calculator, create_llm_calculator
)

# Using factory functions
calc = create_result_calculator()
score = calc.calculate(result_count=15, execution_success=True, used_fallback=False)
# Returns: float between 0.0 and 1.0

# Or from factors dataclass
from src.agents.confidence_calculator import ConfidenceFactors
factors = ConfidenceFactors(result_count=15, execution_success=True)
score = calc.calculate_from_factors(factors)

ConfidenceStrategy (ABC)
  +-- ResultConfidenceStrategy    # Based on result count and execution status
  +-- AnalysisConfidenceStrategy  # Based on intent/domain/keyword analysis
  +-- PatternConfidenceStrategy   # Based on pattern matching success
  +-- LLMConfidenceStrategy       # Based on LLM confidence signals

from src.agents.confidence_calculator import ConfidenceFactors

factors = ConfidenceFactors(
    result_count=15,
    execution_success=True,
    used_fallback=False,
    intent_confidence=0.9,
    domain_confidence=0.85,
    keyword_count=3,
    pattern_match_score=0.75,
    pattern_count=2,
    llm_confidence=0.8,
    coherence_score=0.9
)

from src.agents.prompt_builder import PromptBuilder

builder = PromptBuilder(semantic_prompts: Optional[Dict[str, str]] = None)

# Build enriched prompt with full context
prompt = builder.build_enriched_prompt(question, context)

# Build simple prompt without enrichment
prompt = builder.build_simple_prompt(question, context)

# Build semantic prompt for semantic mode
prompt = builder.build_semantic_prompt(question, context)

# Formatting helpers
enrichment_ctx = builder.format_enrichment_context(hints)
qa_text = builder.format_qa_examples(qa_pairs)
sql_text = builder.format_sql_examples(examples)
guidance = builder.get_domain_guidance(analysis)

from src.agents.prompt_builder import create_prompt_builder

builder = create_prompt_builder(semantic_prompts=None)
# Returns: PromptBuilder instance

from src.agents.query_variants import QueryVariantGenerator

gen = QueryVariantGenerator()

variants = gen.generate_variants(question, context, num_variants=3)
# Returns: [{'query': '...', 'strategy': 'precise'}, {'query': '...', 'strategy': 'broad'}, ...]

# Generate multiple query variants
variants = gen.generate_variants(question, context, num_variants=3)

# Detection helpers for specialized query types
gen.is_data_flow_question("How does data flow from A to B?")   # True
gen.is_control_flow_question("What is the shutdown sequence?")  # True

from src.agents.query_variants import create_variant_generator

gen = create_variant_generator()
# Returns: QueryVariantGenerator instance

from src.agents.fallback_strategies import KeywordToTagMapper

mapper = KeywordToTagMapper(similarity_threshold=None)

from src.agents.fallback_strategies import HybridQueryBuilder

builder = HybridQueryBuilder()

from src.agents.fallback_strategies import FallbackStrategySelector

selector = FallbackStrategySelector()

from src.agents.fallback_strategies import get_fallback_selector

selector = get_fallback_selector()
enhanced_hints = selector.apply_fallback(hints, question, analysis)
# Returns: enhanced hints dict with additional tags from fallback strategies

from src.agents.semantic_interpreter import SemanticInterpreter

interpreter = SemanticInterpreter(
    llm_interface=llm,
    prompt_template=None,               # Custom prompt template
    code_analyst_title="code analyst"   # Role title in prompts
)

result = interpreter.interpret(question, results, query)
# Returns: dict with 'answer', 'confidence', 'sources'

from src.agents.tag_effectiveness_tracker import TagEffectivenessTracker, TagUsageRecord, get_global_tracker

# Singleton access
tracker = get_global_tracker()

# Record tag usage and outcome
tracker.record(tags=["concurrency", "lock-acquire"], success=True, result_count=15)

# Get effectiveness report
report = tracker.get_report()
# Returns: dict mapping tag names to effectiveness scores

# Persistence
tracker.save()
tracker.load()

from src.agents.result_parser import ResultParser

parser = ResultParser()

# Parse function list results
functions = parser.parse_function_list(raw_results)

# Parse semantic results
semantic = parser.parse_semantic_result(raw_results)

# Check for empty results
parser.is_empty(raw_results)

IDE <--> Transport (stdio | HTTP | WebSocket) <--> CodeGraphACPAgent <--> Adapters <--> Core Services

python -m src.acp                    # Default: stdio transport
python -m src.acp --transport http   # HTTP transport on default port
python -m src.acp --transport ws     # WebSocket transport

graph LR
    subgraph "Transport"
        STDIO[StdioTransport]
        HTTP[HTTPTransport]
        WS[WebSocketTransport]
    end
    subgraph "Protocol"
        AGENT[CodeGraphACPAgent]
    end
    subgraph "Server Components"
        SM[SessionManager]
        TM[ThreadManager]
        TO[TurnOrchestrator]
        DE[DiagnosticsEngine]
        HP[HoverProvider]
        AB[ApprovalBridge]
        NR[NotificationRouter]
    end
    subgraph "Integration"
        CA[ChatAdapter]
        SA[SessionAdapter]
        WA[WorkflowAdapter]
        MB[MCPBridge]
    end
    STDIO --> AGENT
    HTTP --> AGENT
    WS --> AGENT
    AGENT --> SM
    AGENT --> TM
    AGENT --> TO
    AGENT --> DE
    AGENT --> HP
    AGENT --> AB
    AGENT --> NR
    AGENT --> CA
    AGENT --> SA
    AGENT --> WA
    AGENT --> MB

from src.acp import CodeGraphACPAgent

agent = CodeGraphACPAgent()  # No parameters

# Main entry point (async)
response = await agent.handle(
    request: JSONRPCRequest,
    notification_callback: Optional[Callable] = None,
    user_id: Optional[str] = None
)

from src.acp import ACPSessionManager

manager = ACPSessionManager(
    max_sessions: int = 100,
    session_timeout_minutes: int = 60
)

# Create a new session
session = await manager.create_session(
    cwd="/path/to/project",
    mcp_servers=[{"id": "...", "transport": "stdio", ...}],
    user_id="user-123"
)

# Load existing session
session = await manager.load_session(session_id="...")

# Get session by ID
session = manager.get_session(session_id="...")

# Cancel a session
await manager.cancel_session(session_id="...")

# Cleanup expired sessions
cleaned = await manager.cleanup_expired()

from src.acp.server.capabilities import CapabilityNegotiator

negotiator = CapabilityNegotiator()

# Negotiate capabilities from initialize params
server_caps = negotiator.negotiate(init_params)
# Returns: AgentCapabilities with negotiated feature flags

# Check if a capability is supported
negotiator.supports("diagnostics")  # True/False
negotiator.supports("hover")        # True/False

# Check if a notification should be sent
negotiator.should_send_notification("diagnostics/publish")  # True/False

from src.acp.server.thread_manager import ACPThreadManager

tm = ACPThreadManager()

# Start a new thread
thread = await tm.start_thread(cwd="/path", name="analysis", config={})

# Resume an existing thread
thread = await tm.resume_thread(thread_id="...")

# Fork a thread at a specific turn
thread = await tm.fork_thread(thread_id="...", at_turn=5)

# List threads with filters
threads = await tm.list_threads(status="active", limit=20, offset=0)

# Archive a thread (preserves but marks inactive)
await tm.archive_thread(thread_id="...")

# Compact thread history to reduce memory
await tm.compact_thread(thread_id="...")

from src.acp.server.turn_orchestrator import TurnOrchestrator

orchestrator = TurnOrchestrator()

# Start a turn (launches background task for streaming)
result = await orchestrator.start_turn(
    thread_id="...", prompt="Analyze this code",
    context={"files": [...]},
    notification_router=router
)

# Interrupt a running turn
result = await orchestrator.interrupt_turn(thread_id="...", turn_id="...")

from src.acp.server.diagnostics import DiagnosticsEngine

engine = DiagnosticsEngine()

# Subscribe to diagnostics for a file
engine.subscribe(uri="file:///path/to/file.py")

# Generate diagnostics for a file
diagnostics = await engine.generate_diagnostics(uri="file:///path/to/file.py")

# Publish diagnostics through notification router
await engine.publish_diagnostics(uri, notification_router=router)

# Unsubscribe from diagnostics
engine.unsubscribe(uri="file:///path/to/file.py")

from src.acp.server.hover import HoverProvider

hover = HoverProvider()

# Get hover information for a position
info = await hover.get_hover_info(
    uri="file:///path/to/file.py",
    line=42,
    character=10
)
# Returns: dict with callers, callees, complexity, taint status from CPG

{
    'symbol': 'process_request',
    'kind': 'method',
    'file': 'handler.py',
    'line': 42,
    'callers': ['main', 'dispatch'],         # Functions that call this symbol
    'callees': ['validate', 'execute'],      # Functions called by this symbol
    'complexity': 8,                          # Cyclomatic complexity
    'taint_status': 'clean',                 # clean | tainted | sanitized
    'documentation': '...'                   # Extracted doc comment if available
}

from src.acp.server.approval_bridge import ACPApprovalBridge

bridge = ACPApprovalBridge()

# Request approval for a file change
result = await bridge.request_file_change_approval(
    thread_id="...", turn_id="...", item_id="...",
    diff="...", files=["file.py"], reason="Refactoring",
    confidence=0.9, notification_router=router, timeout=30.0
)
# Returns: dict with approval decision

# Resolve a pending approval (called when client responds)
bridge.resolve_approval(request_id="...", decision={"approved": True})

# Check pending approvals
bridge.pending_count  # Number of pending approvals

from src.acp.server.notification_router import NotificationRouter

router = NotificationRouter(
    callback=notification_callback,
    negotiator=capability_negotiator
)

# Send a notification
sent = await router.send("session/update", {"sessionUpdate": "plan", ...})
# Returns: bool indicating whether the notification was sent

from src.acp.server.handlers import handle_fs_read, handle_fs_write

# Read a text file
result = await handle_fs_read(
    {"path": "/path/to/file.py"},
    client_capabilities
)
# Returns: {"content": "file contents...", "encoding": "utf-8"}

# Write to a text file
result = await handle_fs_write(
    {"path": "/path/to/file.py", "content": "new content..."},
    client_capabilities
)
# Returns: {"success": True}

from src.acp.server.handlers import (
    handle_terminal_create,
    handle_terminal_output,
    handle_terminal_wait,
    handle_terminal_kill,
    handle_terminal_release
)

# Create a terminal subprocess
result = await handle_terminal_create(
    {"command": "python", "args": ["-m", "pytest"], "cwd": "/project"},
    client_capabilities
)
# Returns: {"terminalId": "..."}

# Read terminal output
result = await handle_terminal_output(
    {"terminalId": "..."},
    client_capabilities
)
# Returns: {"output": "...", "isComplete": False}

# Wait for terminal to exit
result = await handle_terminal_wait(
    {"terminalId": "...", "timeout": 30},
    client_capabilities
)
# Returns: {"exitCode": 0}

# Kill terminal process
result = await handle_terminal_kill(
    {"terminalId": "..."},
    client_capabilities
)
# Returns: {"success": True}

# Release terminal resources
result = await handle_terminal_release(
    {"terminalId": "..."},
    client_capabilities
)
# Returns: {"success": True}

from src.acp.transport.base import BaseTransport  # ABC

class BaseTransport(ABC):
    def __init__(self, agent: Any): ...

    @property
    def is_running(self) -> bool: ...

    async def start(self) -> None: ...      # abstract
    async def stop(self) -> None: ...       # abstract
    async def send(self, message: Any) -> None: ...  # abstract
    async def handle_message(self, message, notification_callback=None, user_id=None) -> Optional[Any]: ...

from src.acp.transport.stdio import StdioTransport, run_stdio

transport = StdioTransport(agent=agent)
await transport.start()  # Blocks, reading from stdin

# Or use the convenience function
await run_stdio()  # Creates agent and transport, runs until EOF

from src.acp.transport.http import HTTPTransport, get_http_transport

transport = HTTPTransport(agent=agent, host="0.0.0.0", port=8080)
await transport.start()

# Or use the factory function
transport = get_http_transport(agent=agent)
await transport.start()

from src.acp.transport.websocket import WebSocketTransport

transport = WebSocketTransport(agent=agent)

from src.acp.transport.websocket import get_websocket_manager

manager = get_websocket_manager()  # Singleton

# Add a connection
await manager.add_connection(user_id="user-123", transport=ws_transport)

# Broadcast to all connections of a user
count = await manager.broadcast_to_user(user_id="user-123", message={"method": "update", ...})

# Get connection count
total = manager.get_connection_count()              # All connections
user_count = manager.get_connection_count(user_id="user-123")  # Specific user

from src.acp.integration.chat_adapter import ChatServiceACPAdapter

adapter = ChatServiceACPAdapter()

# Process a prompt from an ACP session
answer = await adapter.process_prompt(
    content_blocks=[{"type": "text", "text": "Analyze code..."}],
    session=acp_session,
    notification_callback=callback,
    user_id="user-123"
)
# Returns: string answer

from src.acp.integration.session_adapter import SessionACPAdapter

adapter = SessionACPAdapter()

# Create a session
session_id = await adapter.create_session(user_id="user-123", metadata={"ide": "vscode"})

# Get a session
session = await adapter.get_session(session_id="...", user_id="user-123")

# Get session context (recent interactions)
context = await adapter.get_context(session_id="...", user_id="user-123", limit=10)

# Add a dialogue turn
await adapter.add_dialogue_turn(
    session_id="...", user_id="user-123",
    role="user", content="Hello"
)

# Get dialogue history
history = await adapter.get_dialogue_history(session_id="...", user_id="user-123")

from src.acp.integration.workflow_adapter import WorkflowACPAdapter

adapter = WorkflowACPAdapter()

# Get tool kind for a scenario
kind = adapter.get_tool_kind("scenario_1")   # "search", "execute", "analyze", etc.

# Get human-readable tool title
title = adapter.get_tool_title("scenario_1") # e.g., "Code Search"

# Create plan entries for a scenario execution
plan = adapter.create_plan_entries("scenario_1")
# Returns: list of PlanEntry dicts

# Execute workflow with streaming updates
result = await adapter.execute_with_updates(
    query="...", context={}, callback=notification_callback, tool_call_id="tc-1"
)
# Returns: execution result dict

from src.acp.integration.mcp_bridge import MCPBridge

bridge = MCPBridge()

# Connect to MCP servers
connected = await bridge.connect_servers([
    {"id": "server1", "name": "My MCP", "transport": "stdio",
     "command": "node", "args": ["server.js"]},
    {"id": "server2", "name": "Remote MCP", "transport": "http",
     "url": "http://localhost:3000"}
])
# Returns: number of successfully connected servers

# Get all available tools across all servers
tools = bridge.get_all_tools()        # List[MCPTool]

# Get all available resources across all servers
resources = bridge.get_all_resources() # List[MCPResource]

# Call a tool on a specific server
result = await bridge.call_tool("server1", "tool_name", {"arg": "value"})

# Read a resource from a specific server
content = await bridge.read_resource("server1", "resource://uri")

# Disconnect all servers
await bridge.disconnect_all()

from src.acp.cli import setup_logging, main, run

# Functions
setup_logging()     # Configure logging for ACP server
await main()        # Async entry point — parses args and starts transport
run()               # Sync wrapper — calls asyncio.run(main())

from src.security.security_agents import SecurityScanner

scanner = SecurityScanner(cpg_service: Optional[CPGQueryService] = None)

# Context manager support for automatic resource cleanup
with SecurityScanner(cpg_service) as scanner:
    findings = scanner.scan_all_patterns(limit_per_pattern=50)
# Returns: List[SecurityFinding]

from src.security.security_agents import DataFlowAnalyzer

analyzer = DataFlowAnalyzer(cpg_service: Optional[CPGQueryService] = None)

paths = analyzer.trace_taint_flows(limit=100)
# Returns: List[DataFlowPath]

from src.security.security_agents import VulnerabilityReporter

reporter = VulnerabilityReporter()

report = reporter.generate_report(findings, data_flow_paths)
# Returns: VulnerabilityReport

from src.security.security_agents import RemediationAdvisor

advisor = RemediationAdvisor()

advice = advisor.get_remediation_advice(finding)
# Returns: RemediationAdvice

plan = advisor.get_bulk_remediation_plan(findings)
# Returns: List[RemediationAdvice]

from src.performance.agents import PerformanceProfiler

profiler = PerformanceProfiler(cpg_service: Optional[CPGQueryService] = None)

# Context manager support
with PerformanceProfiler(cpg_service) as profiler:
    bottlenecks = profiler.profile_all_bottlenecks(limit_per_pattern=30)
    # Returns: List[BottleneckFinding]

    pattern_findings = profiler.profile_pattern(pattern, limit=30)
    # Returns: List[BottleneckFinding]

from src.performance.agents import ResourceAnalyzer

analyzer = ResourceAnalyzer(cpg_service: Optional[CPGQueryService] = None)

# Context manager support
with ResourceAnalyzer(cpg_service) as analyzer:
    usage = analyzer.analyze_method_resources("method_name", filename="file.py")
# Returns: ResourceUsage