Agent Types

The Health Universe A2A SDK provides two main agent types to fit different use cases and architectural patterns. Understanding when to use each type is crucial for building effective healthcare applications.

Agent vs SubAgent

Agent (AsyncAgent)

The primary agent class for Health Universe applications that handle long-running tasks with progress updates. Agent is an alias for the underlying AsyncAgent implementation.

Key Characteristics:

• Background processing: Tasks run asynchronously with progress updates stored in the database

• SSE acknowledgment: Immediate acknowledgment via Server-Sent Events, then background execution

• Progress updates: Real-time status updates visible in Navigator UI

• Document operations: Full access to document storage and retrieval

• User-facing: Designed for direct user interaction through Navigator

Example Use Cases:

• Clinical data analysis that takes several minutes

• Document processing and extraction workflows

• Machine learning model inference on large datasets

• Multi-step protocol generation

• Report generation with intermediate status updates

from health_universe_a2a import Agent, AgentContext

class ClinicalAnalyzer(Agent):
    def get_agent_name(self) -> str:
        return "Clinical Data Analyzer"
    
    def get_agent_description(self) -> str:
        return "Analyzes clinical datasets and generates comprehensive reports"
    
    async def process_message(self, message: str, context: AgentContext) -> str:
        # Long-running analysis with progress updates
        await context.update_progress("Loading documents...", 0.1)
        docs = await context.document_client.list_documents()
        
        await context.update_progress("Processing data...", 0.5)
        results = await self.analyze_clinical_data(docs)
        
        await context.update_progress("Generating report...", 0.9)
        report = await self.generate_report(results)
        
        # Save results as document
        await context.document_client.write(
            "Analysis Report", 
            report, 
            filename="clinical_analysis.md"
        )
        
        return "Analysis complete! Report generated successfully."

SubAgent

Lightweight agent for fast, inline processing typically called by other agents.

Key Characteristics:

• Inline processing: Returns results directly in HTTP response

• No progress updates: Processing is fast enough that users can wait

• Inter-agent use: Designed to be called by other agents or services

• Simplified context: Uses SubAgentContext with document access but no update mechanisms

• Fast response: Should complete within 30 seconds

Example Use Cases:

• Document format conversion

• Data validation and verification

• Quick calculations or transformations

• Extracting structured data from documents

• Format standardization between systems

from health_universe_a2a import SubAgent, SubAgentContext
import json

class DocumentConverter(SubAgent):
    def get_agent_name(self) -> str:
        return "Document Format Converter"
    
    def get_agent_description(self) -> str:
        return "Converts documents between different medical formats"
    
    async def process_message(self, message: str, context: SubAgentContext) -> str:
        data = json.loads(message)
        
        # Quick format conversion
        doc = await context.document_client.get_document(data["document_id"])
        content = await context.document_client.download_text(doc.id)
        
        # Fast processing - no progress updates needed
        converted = await self.convert_format(content, data["target_format"])
        
        return json.dumps({
            "status": "success",
            "converted_content": converted,
            "format": data["target_format"]
        })

Comparison Table

Feature	Agent (AsyncAgent)	SubAgent
Processing Style	Background/async	Inline/direct
Response Time	Minutes to hours	Seconds to ~30s
Progress Updates	✅ Real-time updates	❌ No updates
User Interface	Navigator with progress bar	Called by other agents
Document Access	✅ Full operations	✅ Read/write access
Inter-agent Calls	✅ Can call other agents	✅ Can call other agents
Background Jobs	✅ Database-persisted	❌ Not applicable
Context Type	AgentContext (BackgroundContext)	SubAgentContext
Streaming Support	✅ SSE acknowledgment	❌ Direct response

Architectural Patterns

Pattern 1: Orchestrator + Workers

Use an Agent as orchestrator with SubAgents as workers:

# Orchestrator (Agent/AsyncAgent)
class ProtocolOrchestrator(Agent):
    async def process_message(self, message: str, context: AgentContext) -> str:
        await context.update_progress("Starting protocol generation...", 0.1)
        
        # Call SubAgent for document analysis
        analysis = await self.call_agent(
            "document-analyzer", 
            message, 
            context
        )
        
        await context.update_progress("Generating sections...", 0.5)
        
        # Call SubAgent for section writing
        sections = await self.call_agent(
            "section-writer",
            json.dumps(analysis),
            context
        )
        
        await context.update_progress("Finalizing protocol...", 0.9)
        return "Protocol generation complete!"

# Worker (SubAgent)  
class DocumentAnalyzer(SubAgent):
    async def process_message(self, message: str, context: SubAgentContext) -> str:
        # Fast document analysis
        docs = await context.document_client.list_documents()
        analysis = await self.analyze_documents(docs)
        return json.dumps(analysis)

Pattern 2: Multi-Agent Workflow

Chain multiple Agents for complex workflows:

# Deploy multiple agents at different paths
from health_universe_a2a import serve_multi_agents

serve_multi_agents({
    "/orchestrator": ProtocolOrchestrator(),
    "/document-processor": DocumentProcessor(),  # Agent (AsyncAgent)
    "/section-writer": SectionWriter(),          # Agent (AsyncAgent)
    "/format-converter": FormatConverter(),      # SubAgent
})

When to Choose Each Type

Choose Agent when:

• ✅ Processing takes more than 30 seconds

• ✅ Users need progress visibility

• ✅ Working with large datasets or documents

• ✅ Running machine learning inference

• ✅ Users interact directly through Navigator

• ✅ Need to generate substantial outputs or reports

• ✅ Process involves multiple steps that benefit from status updates

Choose SubAgent when:

• ✅ Processing completes in under 30 seconds

• ✅ Called by other agents rather than users directly

• ✅ Performing simple transformations or validations

• ✅ Converting between data formats

• ✅ Extracting specific information from documents

• ✅ Acting as a microservice in larger workflows

• ✅ Providing utility functions to other agents

Best Practices

For Agents:

• Send meaningful progress updates every 10-20% of completion

• Use descriptive status messages that help users understand progress

• Handle long-running operations with proper error recovery

• Save intermediate results to documents when appropriate

• Implement cancellation checks for very long processes

For SubAgents:

• Keep processing under 30 seconds to avoid timeouts

• Return structured JSON for easy consumption by calling agents

• Focus on single, well-defined responsibilities

• Use clear input/output schemas for inter-agent communication

• Handle errors gracefully with informative error messages

For Both:

• Use descriptive agent names and descriptions for Navigator discovery

• Implement proper error handling and logging

• Follow healthcare data security best practices

• Test thoroughly with realistic data volumes

• Document expected inputs and outputs clearly

The choice between Agent and SubAgent should be driven by your use case requirements, expected processing time, and whether users need visibility into progress. Many sophisticated healthcare applications use both types working together in complementary roles.