health-universe-a2a 0.2.0
Python SDK for building Health Universe A2A-compliant agents
License: MIT
Requires Python: >=3.10
Keywords: a2a, agents, ai, health-universe

# Health Universe A2A SDK for Python

[![Test](https://github.com/Health-Universe/healthuniverse-a2a-sdk-python/actions/workflows/test.yml/badge.svg)](https://github.com/Health-Universe/healthuniverse-a2a-sdk-python/actions/workflows/test.yml)

A simple, batteries-included Python SDK for building [A2A-compliant agents](https://a2a.ai) 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:

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

For development:

```bash
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](https://github.com/astral-sh/uv) is recommended for faster, more reliable dependency management. You can also use `pip install git+...` if preferred.

## Quick Start

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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](examples/simple_agent.py)**: Basic echo agent
- **[medical_classifier.py](examples/medical_classifier.py)**: Simple symptom classifier
- **[document_inventory.py](examples/document_inventory.py)**: List and inspect thread documents
- **[protocol_analyzer.py](examples/protocol_analyzer.py)**: Search, download, and analyze documents
- **[physician_followup_agent.py](examples/physician_followup_agent.py)**: SOAP note analysis with OpenAI
- **[multi_agent_orchestration.py](examples/multi_agent_orchestration.py)**: Multiple agents working together

## Inter-Agent Communication

Call other A2A-compliant agents from your agent:

```python
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

```python
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

```python
# 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:

```python
from health_universe_a2a import serve_multi_agents

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

## Development

### Setup

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

### Testing

```bash
uv run pytest
```

### Linting and Formatting

```bash
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](https://docs.healthuniverse.com/a2a-sdk)
- [Issue Tracker](https://github.com/Health-Universe/healthuniverse-a2a-sdk-python/issues)
- [Email Support](mailto:support@healthuniverse.com)

## Related Projects

- [Health Universe Platform](https://healthuniverse.com)
- [A2A Protocol Specification](https://a2a.ai)