Agent Lifecycle

from solace_agent_mesh.agent.sac.app import SamAgentApp

# Configuration is loaded automatically from YAML
app = SamAgentApp(app_config)

# These are handled automatically by the framework
component.session_service  # Initialized session service
component.artifact_service  # Initialized artifact service
component.adk_agent  # ADK agent instance

# Built-in tools are automatically registered
# MCP tools are loaded from configured servers
# Custom tools can be registered programmatically

Topic: a2a/<namespace>/agent/<agent_name>/request

class TaskExecutionContext:
    task_id: str
    session: ADKSession
    a2a_context: Dict[str, Any]
    is_paused: bool
    produced_artifacts: Dict[str, int]
    active_peer_sub_tasks: Dict[str, Any]

async def run_adk_async_task(
    component: SamAgentComponent,
    task_context: TaskExecutionContext,
    adk_session: ADKSession,
    adk_content: Content,
    run_config: RunConfig,
    a2a_context: Dict[str, Any],
) -> bool:
    """
    Runs the ADK agent asynchronously.
    
    Returns:
        bool: True if task is paused for long-running operation
    """
    async for event in component.runner.run_async(
        session=adk_session,
        content=adk_content,
        config=run_config,
    ):
        # Process streaming events
        await process_event(event)

def repair_history_callback(
    callback_context: CallbackContext,
    llm_request: LlmRequest
) -> Optional[LlmResponse]:
    """
    Called before sending request to LLM.
    Can modify history or inject context.
    """
    # Repair any dangling tool calls
    # Add dynamic instructions
    return None  # Continue to LLM

async def process_artifact_blocks_callback(
    callback_context: CallbackContext,
    llm_response: LlmResponse,
    host_component: SamAgentComponent,
) -> Optional[LlmResponse]:
    """
    Called after receiving LLM response.
    Processes artifact blocks and streaming content.
    """
    # Parse fenced artifact blocks
    # Save artifacts
    # Stream to client
    return None  # Return modified response

async def manage_large_mcp_tool_responses_callback(
    tool: BaseTool,
    args: Dict[str, Any],
    tool_context: ToolContext,
    tool_response: Any,
    host_component: SamAgentComponent,
) -> Optional[Dict[str, Any]]:
    """
    Called after tool execution.
    Handles large responses and artifact creation.
    """
    if response_too_large(tool_response):
        artifact = save_as_artifact(tool_response)
        return {"artifact_reference": artifact}
    return tool_response

async def _create_compaction_event(
    component: SamAgentComponent,
    session: ADKSession,
    compaction_threshold: float = 0.25,
) -> tuple[int, str]:
    """
    Summarizes oldest 25% of conversation history.
    Returns (events_compacted, summary_text)
    """
    # Calculate target compaction size
    # Find user turn boundary
    # Use LLM to summarize events
    # Persist compaction event
    # Reload session with filtered events

Turn 1-10 → Summary A (1000 tokens)
Turn 11-20 → Summary B (1000 tokens) = Summarize(Summary A + Turn 11-20)
Turn 21-30 → Summary C (1000 tokens) = Summarize(Summary B + Turn 21-30)

async def finalize_task_with_cleanup(
    component: SamAgentComponent,
    a2a_context: Dict[str, Any],
    is_paused: bool,
    exception: Optional[Exception] = None,
):
    """
    Finalize task execution and cleanup resources.
    """
    logical_task_id = a2a_context.get("logical_task_id")
    
    # Flush streaming buffer
    task_context.flush_streaming_buffer()
    
    # Send final status update
    if exception:
        send_error_response(exception)
    else:
        send_success_response()
    
    # Cleanup task context
    component.active_tasks.pop(logical_task_id)

class SamAgentComponent:
    async def shutdown(self):
        """
        Graceful shutdown sequence.
        """
        # Stop accepting new requests
        self.shutdown_event.set()
        
        # Wait for active tasks to complete (with timeout)
        await self._wait_for_active_tasks(timeout=30)
        
        # Cancel remaining tasks
        await self._cancel_remaining_tasks()
        
        # Cleanup resources
        await self.session_service.shutdown()
        await self.artifact_service.shutdown()
        
        # Close broker connection
        await self.broker_connection.disconnect()

try:
    response = await adk_agent.run(content)
except BadRequestError as e:
    if _is_context_limit_error(e):
        # Automatic retry with summarization
        await _create_compaction_event(session)
        reloaded_session = await session_service.get_session(session.id)
        response = await adk_agent.run(content)  # Retry

except TaskCancelledError as e:
    # Propagate to peer sub-tasks
    for sub_task_id in task_context.active_peer_sub_tasks:
        send_cancel_request(sub_task_id)
    
    # Send cancellation response
    send_task_cancelled_response()

except LlmCallsLimitExceededError as e:
    # Max iterations reached
    send_error_response(
        "Maximum LLM call limit exceeded. Task may be too complex."
    )

session_service:
  type: "database"
  default_behavior: "PERSISTENT"

session_service:
  type: "memory"
  default_behavior: "RUN_BASED"

component.active_tasks: Dict[str, TaskExecutionContext]
component.active_tasks_lock: asyncio.Lock

# Stream partial responses to client
for chunk in llm_response.stream():
    await send_streaming_chunk(chunk)

try:
    await execute_task()
finally:
    # Always cleanup
    task_context.cleanup()
    component.active_tasks.pop(task_id)

log.info("%s Starting task %s", component.log_identifier, task_id)
log.debug("%s Session events: %d", log_identifier, len(session.events))
log.warning("%s Context limit reached, compacting...", log_identifier)

# Check active tasks
with component.active_tasks_lock:
    task_context = component.active_tasks.get(task_id)
    print(f"Task paused: {task_context.is_paused}")
    print(f"Produced artifacts: {task_context.produced_artifacts}")