Local Development

When developing A2A agents, it's essential to test your agent locally before deploying to Health Universe. The SDK provides local development tools that simulate the production environment without requiring backend services.

This guide shows you how to set up and test your agents locally using the LocalDocumentClient and create_local_context.

Overview

Local development with A2A agents allows you to:

• Test agent logic without deploying to Health Universe

• Work with local test files instead of cloud storage

• Debug your agent's document processing capabilities

• Validate input/output behavior in isolation

The SDK provides two key utilities for local testing:

• LocalDocumentClient: Filesystem-backed document operations

• create_local_context: Factory for creating test contexts

Setting Up Local Test Data

Directory Structure

Create a local test data directory with the following structure:

test_data/
├── source/          # Input documents (user uploads)
│   ├── protocol.pdf
│   ├── data.csv
│   └── notes.txt
└── artifact/        # Pre-seeded outputs (optional)
    └── example.json

The SDK will automatically create these directories if they don't exist.

Supported File Types

The LocalDocumentClient works with any file type, but for realistic testing, use files similar to what users will upload:

• PDFs: Clinical protocols, research papers

• CSV: Patient data, lab results

• JSON: Structured medical data

• TXT/MD: Clinical notes, documentation

Creating a Local Context

The simplest way to create a local test environment is using create_local_context():

from health_universe_a2a.local import create_local_context

# Create context with default output directory
context = create_local_context(data_dir="./test_data")

# Or specify a custom output directory
context = create_local_context(
    data_dir="./test_data",
    output_dir="./output"
)

Testing Your Agent Locally

Basic Local Testing

Here's a complete example of testing an agent locally:

import asyncio
import json
from health_universe_a2a import Agent, AgentContext
from health_universe_a2a.local import create_local_context

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

    def get_agent_description(self) -> str:
        return "Analyzes uploaded documents and generates insights"

    async def process_message(self, message: str, context: AgentContext) -> str:
        # List all source documents
        docs = await context.document_client.list_documents(role="source")
        print(f"Found {len(docs)} source documents")

        results = []
        for doc in docs:
            # Read document content (works for text files)
            try:
                content = await context.document_client.download_text(doc.id)
                word_count = len(content.split())
                results.append({
                    "document": doc.name,
                    "word_count": word_count,
                    "size_bytes": len(content.encode())
                })
                
                # Update progress
                await context.update_progress(
                    f"Analyzed {doc.name}",
                    progress=len(results) / len(docs)
                )
            except UnicodeDecodeError:
                # Handle binary files
                content_bytes = await context.document_client.download(doc.id)
                results.append({
                    "document": doc.name,
                    "type": "binary",
                    "size_bytes": len(content_bytes)
                })

        # Write results to output
        await context.document_client.write(
            "Analysis Results",
            json.dumps(results, indent=2),
            filename="analysis.json"
        )

        return f"Analyzed {len(docs)} documents. Results saved to analysis.json"

async def test_locally():
    # Create agent and local context
    agent = DocumentAnalyzer()
    context = create_local_context("./test_data")

    # Test the agent
    result = await agent.process_message("Analyze documents", context)
    print(f"Result: {result}")

if __name__ == "__main__":
    asyncio.run(test_locally())

Advanced Local Testing

For more sophisticated testing, you can create the LocalDocumentClient directly:

from health_universe_a2a.local import LocalDocumentClient
from health_universe_a2a import BackgroundContext

async def advanced_local_test():
    # Create document client
    doc_client = LocalDocumentClient(
        data_dir="./test_data",
        output_dir="./output"
    )

    # Build context with local document client
    context = BackgroundContext(
        user_id="test-user",
        thread_id="test-thread",
        job_id="test-job-123"
    )
    # Override with local document client
    context._document_client_override = doc_client

    # Test document operations
    docs = await doc_client.list_documents()
    print(f"Available documents: {[d.name for d in docs]}")

    # Filter by name
    protocols = await doc_client.filter_by_name("protocol")
    print(f"Protocol documents: {[d.name for d in protocols]}")

if __name__ == "__main__":
    asyncio.run(advanced_local_test())

Working with Different File Types

Text Files

For text-based files (CSV, JSON, TXT, MD), use download_text():

# Read CSV data
csv_docs = await context.document_client.filter_by_name("data.csv")
if csv_docs:
    csv_content = await context.document_client.download_text(csv_docs[0].id)
    lines = csv_content.split('\n')
    print(f"CSV has {len(lines)} rows")

Binary Files

For binary files (PDF, images, etc.), use download():

# Read PDF bytes
pdf_docs = await context.document_client.filter_by_name("protocol.pdf")
if pdf_docs:
    pdf_bytes = await context.document_client.download(pdf_docs[0].id)
    print(f"PDF size: {len(pdf_bytes)} bytes")
    
    # Save to local file for inspection
    with open("downloaded_protocol.pdf", "wb") as f:
        f.write(pdf_bytes)

Creating Test Output

Write various output formats to test different scenarios:

# JSON output
await context.document_client.write(
    "Structured Results",
    json.dumps({"score": 0.95, "status": "complete"}),
    filename="results.json"
)

# Markdown report
report = """
# Analysis Report

## Summary
Document analysis completed successfully.

## Findings
- 3 documents processed
- Average word count: 1,250
- No errors detected
"""

await context.document_client.write(
    "Analysis Report",
    report,
    filename="report.md"
)

# CSV output
csv_data = "document,words,score\nprotocol.pdf,1500,0.95\ndata.csv,50,0.80"
await context.document_client.write(
    "Document Scores",
    csv_data,
    filename="scores.csv"
)

Limitations of Local Mode

When testing locally, be aware of these limitations:

Search Not Available

Full-text and semantic search are not available in local mode:

# This will raise NotImplementedError in local mode
try:
    results = await context.document_client.search("metformin")
except NotImplementedError:
    print("Search not available in local mode")

# This will also raise NotImplementedError
try:
    results = await context.document_client.semantic_search("drug interactions")
except NotImplementedError:
    print("Semantic search not available in local mode")

No Text Extraction Pipeline

Local mode doesn't have the platform's text extraction pipeline:

# In production, this extracts text from PDFs
# In local mode, it falls back to download_text() (will fail for PDFs)
try:
    extracted = await context.document_client.download_extracted(pdf_doc.id)
except UnicodeDecodeError:
    print("Text extraction not available locally for binary files")

Progress Updates Are Logged Only

Progress updates are logged to stdout instead of being sent to the backend:

# This will print to console in local mode
await context.update_progress("Processing...", 0.5)
# Output: [LOCAL] Processing... (50.0%)

Best Practices

Organize Test Data

Keep your test data organized and representative:

# Create different test scenarios
test_scenarios = [
    "test_data/clinical_trial",      # Clinical trial documents
    "test_data/patient_records",     # Patient data files  
    "test_data/lab_results",         # Laboratory data
    "test_data/edge_cases"           # Error/edge case files
]

for scenario in test_scenarios:
    print(f"\nTesting scenario: {scenario}")
    context = create_local_context(scenario)
    result = await agent.process_message("test", context)
    print(f"Result: {result}")

Test Error Conditions

Include problematic files to test error handling:

async def test_error_handling():
    # Test with empty directory
    context = create_local_context("./empty_data")
    result = await agent.process_message("test empty", context)
    
    # Test with corrupted files
    context = create_local_context("./corrupted_data") 
    result = await agent.process_message("test corrupted", context)

Validate Output Structure

Check that your agent produces the expected output format:

# After running agent locally
output_docs = await context.document_client.list_documents(role="artifact")
assert len(output_docs) > 0, "Agent should produce output documents"

# Verify specific output file
results_docs = await context.document_client.filter_by_name("results.json")
assert len(results_docs) == 1, "Should produce exactly one results file"

results_content = await context.document_client.download_text(results_docs[0].id)
results_data = json.loads(results_content)
assert "score" in results_data, "Results should include score field"

Moving from Local to Production

Once your agent works locally, deploying to production should be seamless:

1. Keep the same agent code - no changes needed

2. Replace local context with production AgentContext

3. Remove local test code - the production environment handles context creation

4. Test with real data - upload test files to your Health Universe thread

The context.document_client API works identically in both local and production modes, so your document processing code remains the same.

Example: Complete Local Development Workflow

Here's a complete example showing the full local development workflow:

#!/usr/bin/env python3
"""
Local development script for Document Analyzer agent
"""

import asyncio
import json
import logging
from pathlib import Path
from health_universe_a2a import Agent, AgentContext
from health_universe_a2a.local import create_local_context

# Set up logging
logging.basicConfig(level=logging.INFO)

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

    def get_agent_description(self) -> str:
        return "Analyzes documents and generates insights"

    async def process_message(self, message: str, context: AgentContext) -> str:
        # Your agent logic here
        docs = await context.document_client.list_documents(role="source")
        
        if not docs:
            return "No documents found to analyze"
        
        # Process documents
        results = []
        for i, doc in enumerate(docs):
            await context.update_progress(
                f"Analyzing {doc.name}...", 
                progress=(i + 1) / len(docs)
            )
            
            # Analyze document (simplified example)
            try:
                content = await context.document_client.download_text(doc.id)
                results.append({
                    "name": doc.name,
                    "word_count": len(content.split()),
                    "char_count": len(content)
                })
            except UnicodeDecodeError:
                # Binary file
                content = await context.document_client.download(doc.id)
                results.append({
                    "name": doc.name,
                    "type": "binary",
                    "size": len(content)
                })
        
        # Save results
        await context.document_client.write(
            "Analysis Results",
            json.dumps(results, indent=2),
            filename="analysis_results.json"
        )
        
        return f"Analyzed {len(docs)} documents successfully"

async def main():
    """Run local development testing"""
    
    # Setup test data
    test_dir = Path("./test_data")
    test_dir.mkdir(exist_ok=True)
    (test_dir / "source").mkdir(exist_ok=True)
    
    # Create sample test file if none exist
    source_dir = test_dir / "source"
    if not any(source_dir.iterdir()):
        (source_dir / "sample.txt").write_text(
            "This is a sample document for testing the Document Analyzer agent."
        )
        print("Created sample test file: test_data/source/sample.txt")
    
    # Initialize agent and context
    agent = DocumentAnalyzer()
    context = create_local_context(str(test_dir))
    
    print(f"Testing {agent.get_agent_name()}")
    print(f"Description: {agent.get_agent_description()}")
    print("-" * 50)
    
    # Test the agent
    result = await agent.process_message("Analyze all documents", context)
    
    print(f"\nAgent Result: {result}")
    
    # Show output files
    output_docs = await context.document_client.list_documents(role="artifact")
    if output_docs:
        print(f"\nGenerated {len(output_docs)} output file(s):")
        for doc in output_docs:
            print(f"  - {doc.name} ({doc.filename})")
    
    print("\nLocal testing complete!")

if __name__ == "__main__":
    asyncio.run(main())

Save this as test_local.py and run it to test your agent locally before deployment.

This local development approach lets you iterate quickly, debug issues, and ensure your agent works correctly before deploying to Health Universe.