health-universe-a2a 0.2.0

Python SDK for building Health Universe A2A-compliant agents

Health Universe A2A SDK for Python

A simple, batteries-included Python SDK for building A2A-compliant agents for the Health Universe platform.

Features

• Simple API: Just implement 3 methods to create an agent
• Document Operations: Built-in support for reading and writing files
• Multi-Agent Support: Run multiple agents with inter-agent communication
• Progress Updates: Built-in support for progress tracking and artifacts
• Validation: Pre-validate messages before processing
• Lifecycle Hooks: Customize behavior at key points
• Health Universe Integration: Works seamlessly with HU platform

Installation

Install directly from the public GitHub repository:

uv pip install git+https://github.com/Health-Universe/healthuniverse-a2a-sdk-python.git

For development:

git clone https://github.com/Health-Universe/healthuniverse-a2a-sdk-python.git
cd healthuniverse-a2a-sdk-python
uv pip install -e ".[dev]"

Note: Using uv is recommended for faster, more reliable dependency management. You can also use pip install git+... if preferred.

Quick Start

from health_universe_a2a import Agent, AgentContext

class SymptomClassifierAgent(Agent):
    def get_agent_name(self) -> str:
        return "Symptom Classifier"

    def get_agent_description(self) -> str:
        return "Classifies symptoms into medical categories"

    async def process_message(self, message: str, context: AgentContext) -> str:
        await context.update_progress("Analyzing symptoms...", 0.5)

        # Your classification logic here
        category = classify_symptoms(message)

        return f"Classification: {category}"

if __name__ == "__main__":
    SymptomClassifierAgent().serve()

Working with Documents

The SDK provides a DocumentClient for reading and writing files in the Health Universe platform:

from health_universe_a2a import Agent, AgentContext

class DocumentAnalyzerAgent(Agent):
    def get_agent_name(self) -> str:
        return "Document Analyzer"

    def get_agent_description(self) -> str:
        return "Analyzes clinical documents"

    async def process_message(self, message: str, context: AgentContext) -> str:
        # List all documents in the thread
        documents = await context.document_client.list_documents()

        # Filter documents by name
        protocols = await context.document_client.filter_by_name("protocol")

        # Download and read a document
        content = await context.document_client.download_text(documents[0].id)

        # Write results back
        await context.document_client.write(
            name="Analysis Results",
            content='{"result": "analysis complete"}',
            filename="analysis_results.json",
        )

        return f"Analyzed {len(documents)} documents"

Core Concepts

Agent Context

Your process_message method receives an AgentContext with helper methods:

async def process_message(self, message: str, context: AgentContext) -> str:
    # Send progress updates
    await context.update_progress("Working...", 0.5)

    # Add artifacts (files generated by the agent)
    # Prefer markdown format - the platform has markdown WYSIWYG support
    await context.add_artifact(
        name="Results",
        content=markdown_report,
        data_type="text/markdown"
    )

    # Access metadata
    user_id = context.user_id
    thread_id = context.thread_id

    # Access documents API
    docs = await context.document_client.list_documents()

    return "Done!"

Note: Automatic Terminal Status

The SDK automatically sends a terminal status (completed or failed) when process_message() returns or raises an exception. This ensures the Navigator progress bar always completes properly.

Tip: Prefer Markdown for Artifacts

When generating artifacts, prefer text/markdown as the data type. The Health Universe platform includes a markdown WYSIWYG editor, so users can view and edit markdown artifacts directly in the browser.

Validation

Validate messages before processing:

from health_universe_a2a import ValidationAccepted, ValidationRejected

async def validate_message(self, message: str, metadata: dict) -> ValidationAccepted | ValidationRejected:
    if len(message) < 10:
        return ValidationRejected(reason="Message too short (min 10 chars)")

    return ValidationAccepted(estimated_duration_seconds=60)

Lifecycle Hooks

Customize behavior at key points:

async def on_startup(self) -> None:
    """Called when agent starts up"""
    self.model = await load_model()

async def on_shutdown(self) -> None:
    """Called when agent shuts down"""
    await self.model.unload()

async def on_task_start(self, message: str, context: AgentContext) -> None:
    """Called before processing"""
    self.logger.info(f"Starting task for {context.user_id}")

async def on_task_complete(self, message: str, result: str, context: AgentContext) -> None:
    """Called after successful processing"""
    await self.metrics.increment("tasks_completed")

async def on_task_error(self, message: str, error: Exception, context: AgentContext) -> str | None:
    """Called on error - return custom error message or None for default"""
    if isinstance(error, TimeoutError):
        return "Task timed out. Try a smaller request."
    return None

Configuration Methods

Customize agent behavior:

def get_agent_version(self) -> str:
    """Version string (default: "1.0.0")"""
    return "2.1.0"

def get_max_duration_seconds(self) -> int:
    """Max duration hint (default: 3600)"""
    return 7200  # 2 hours

def get_supported_input_formats(self) -> list[str]:
    """Supported input MIME types"""
    return ["text/plain", "application/json"]

def get_supported_output_formats(self) -> list[str]:
    """Supported output MIME types"""
    return ["text/plain", "application/json"]

Examples

See the examples/ directory for complete working examples:

• simple_agent.py: Basic echo agent
• medical_classifier.py: Simple symptom classifier
• document_inventory.py: List and inspect thread documents
• protocol_analyzer.py: Search, download, and analyze documents
• physician_followup_agent.py: SOAP note analysis with OpenAI
• multi_agent_orchestration.py: Multiple agents working together

Inter-Agent Communication

Call other A2A-compliant agents from your agent:

from health_universe_a2a import Agent, AgentContext

class OrchestratorAgent(Agent):
    async def process_message(self, message: str, context: AgentContext) -> str:
        # Call with text message
        preprocessor_result = await self.call_agent(
            "/preprocessor",
            message,
            context,
        )

        # Call with structured data (dict or list)
        analysis_result = await self.call_agent(
            "/analyzer",
            {"data": preprocessor_result, "mode": "detailed"},
            context,
        )

        return analysis_result

Agent Identifier Formats:

1. Local agent path: /agent-name - Uses LOCAL_AGENT_BASE_URL (default: http://localhost:8501)
2. Direct URL: https://... - Calls directly with HTTPS
3. Registry name: agent-name - Looks up in AGENT_REGISTRY environment variable

Running Your Agent

Built-in HTTP Server

if __name__ == "__main__":
    agent = MyAgent()
    agent.serve()  # Starts server on http://0.0.0.0:8000

The server automatically provides:

• Agent card endpoint: GET /.well-known/agent-card.json
• JSON-RPC endpoint: POST / (method: "message/send")
• Health check: GET /health

Server Configuration

# Via environment variables
# HOST=0.0.0.0 PORT=8080 python my_agent.py

# Via method parameters
agent.serve(host="0.0.0.0", port=8080, reload=True)

Multi-Agent Server

Run multiple agents in a single server:

from health_universe_a2a import serve_multi_agents

serve_multi_agents({
    "/orchestrator": OrchestratorAgent(),
    "/analyzer": AnalyzerAgent(),
    "/reader": ReaderAgent(),
}, port=8501)

Development

Setup

git clone https://github.com/Health-Universe/healthuniverse-a2a-sdk-python
cd healthuniverse-a2a-sdk-python
uv pip install -e ".[dev]"

Testing

uv run pytest

Linting and Formatting

uv run ruff check src/
uv run ruff format src/
uv run mypy src/

Requirements

• Python 3.10+
• httpx >= 0.27.0
• pydantic >= 2.0.0
• openai >= 1.0.0

Support

• Documentation
• Issue Tracker
• Email Support

Related Projects

• Health Universe Platform
• A2A Protocol Specification