Cross-Language Collaboration: Python + Go Agent Collaboration

User request
    │
    ▼
┌─────────────────────────────────────────────────────────────┐
│               Go Agent (API Gateway)                        │
│        Responsibilities: routing, auth, rate limiting       │
│              Stack: Go + ADK + A2A Client                   │
└────────┬────────────────────────────┬───────────────────────┘
         │                            │
         ▼                            ▼
┌──────────────────┐          ┌──────────────────┐
│ Python Agent      │          │ Go Agent          │
│ (Data Processing) │          │ (Report Generator)│
│  Responsibilities:│          │  Responsibilities:│
│   data cleaning,  │          │   formatting,     │
│   statistics      │          │   file export     │
│  Stack: Python +  │          │  Stack: Go + ADK  │
│   Pandas +        │          │                    │
│   NumPy           │          │                    │
└────────┬──────────┘          └────────┬──────────┘
         │                              │
         ▼                              ▼
┌──────────────────┐          ┌──────────────────┐
│ Data Storage      │          │ File Storage      │
│ (PostgreSQL)      │          │ (S3 / MinIO)      │
└──────────────────┘          └──────────────────┘

1. User: "Analyze sales_data.csv and generate the quarterly report"
         │
         ▼
2. Go Gateway Agent
   - Verify user permissions
   - Parse the request intent
   - Route to the Python Data Agent
         │
         ▼
3. Python Data Agent
   - Download sales_data.csv from S3
   - Clean the data with Pandas
   - Compute quarterly statistics
   - Store results in PostgreSQL
   - Return a summary of the processing results
         │
         ▼
4. Go Gateway Agent
   - Receive the data processing result
   - Call the Go Report Agent
         │
         ▼
5. Go Report Agent
   - Read statistics from PostgreSQL
   - Generate a PDF report
   - Upload to S3
   - Return the report download link
         │
         ▼
6. Go Gateway Agent
   - Aggregate all results
   - Return them to the user

python-data-agent/
├── Dockerfile
├── requirements.txt
├── src/
│   ├── __init__.py
│   ├── agent.py          # Agent core logic
│   ├── data_processor.py  # Data processing module
│   ├── storage.py        # Storage interface
│   └── a2a_server.py     # A2A server
├── config/
│   └── config.yaml
└── tests/
    └── test_processor.py

# src/agent.py
from dataclasses import dataclass
from typing import Dict, List, Optional, Any
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)

@dataclass
class AnalysisResult:
    total_records: int
    date_range: tuple
    revenue_stats: Dict[str, float]
    top_products: List[Dict[str, Any]]
    monthly_trend: List[Dict[str, Any]]
    anomalies: List[Dict[str, Any]]
    raw_data_path: str

class DataProcessor:
    """Core data processing class"""

    def __init__(self, storage_client):
        self.storage = storage_client
        self.required_columns = ['date', 'product', 'quantity', 'price', 'revenue']

    def process(self, file_path: str, analysis_type: str = 'full') -> AnalysisResult:
        """Process a data file and return the analysis result"""

        # 1. Read the data
        logger.info(f"Reading data from {file_path}")
        df = self._read_data(file_path)

        # 2. Clean the data
        logger.info("Cleaning data")
        df = self._clean_data(df)

        # 3. Run different processing based on the analysis type
        if analysis_type == 'full':
            result = self._full_analysis(df)
        elif analysis_type == 'summary':
            result = self._summary_analysis(df)
        else:
            raise ValueError(f"Unknown analysis type: {analysis_type}")

        # 4. Store the processed data in the database
        db_path = self.storage.save_dataframe(df, f"processed_{datetime.now().strftime('%Y%m%d_%H%M%S')}")
        result.raw_data_path = db_path

        return result

    def _read_data(self, file_path: str) -> pd.DataFrame:
        """Read a CSV/Excel file"""
        if file_path.endswith('.csv'):
            df = pd.read_csv(file_path)
        elif file_path.endswith(('.xlsx', '.xls')):
            df = pd.read_excel(file_path)
        else:
            raise ValueError(f"Unsupported file format: {file_path}")

        # Validate required columns
        missing = set(self.required_columns) - set(df.columns)
        if missing:
            raise ValueError(f"Missing required columns: {missing}")

        return df

    def _clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """Data cleaning"""
        # Remove duplicate rows
        df = df.drop_duplicates()

        # Handle missing values
        df = df.dropna(subset=['date', 'product', 'revenue'])

        # Convert the date format
        df['date'] = pd.to_datetime(df['date'])

        # Remove outliers (revenue < 0 or > 99.9 percentile)
        q99 = df['revenue'].quantile(0.999)
        df = df[(df['revenue'] >= 0) & (df['revenue'] <= q99)]

        # Sort
        df = df.sort_values('date')

        logger.info(f"Cleaned data: {len(df)} records")
        return df

    def _full_analysis(self, df: pd.DataFrame) -> AnalysisResult:
        """Full analysis"""

        # Basic statistics
        total_records = len(df)
        date_range = (df['date'].min().isoformat(), df['date'].max().isoformat())

        # Revenue statistics
        revenue_stats = {
            'total': float(df['revenue'].sum()),
            'mean': float(df['revenue'].mean()),
            'median': float(df['revenue'].median()),
            'std': float(df['revenue'].std()),
            'min': float(df['revenue'].min()),
            'max': float(df['revenue'].max()),
        }

        # Top 10 selling products
        top_products = df.groupby('product').agg({
            'revenue': 'sum',
            'quantity': 'sum'
        }).sort_values('revenue', ascending=False).head(10)

        top_products_list = [
            {
                'product': idx,
                'revenue': float(row['revenue']),
                'quantity': int(row['quantity'])
            }
            for idx, row in top_products.iterrows()
        ]

        # Monthly trend
        df['month'] = df['date'].dt.to_period('M')
        monthly = df.groupby('month')['revenue'].sum().reset_index()
        monthly_trend = [
            {
                'month': str(row['month']),
                'revenue': float(row['revenue'])
            }
            for _, row in monthly.iterrows()
        ]

        # Anomaly detection (3-sigma rule)
        mean = df['revenue'].mean()
        std = df['revenue'].std()
        anomalies = df[abs(df['revenue'] - mean) > 3 * std]
        anomalies_list = [
            {
                'date': row['date'].isoformat(),
                'product': row['product'],
                'revenue': float(row['revenue']),
                'deviation': float(abs(row['revenue'] - mean) / std)
            }
            for _, row in anomalies.iterrows()
        ]

        return AnalysisResult(
            total_records=total_records,
            date_range=date_range,
            revenue_stats=revenue_stats,
            top_products=top_products_list,
            monthly_trend=monthly_trend,
            anomalies=anomalies_list,
            raw_data_path=""
        )

# src/a2a_server.py
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, Dict, Any
import uvicorn
import os
import logging

from agent import DataProcessor
from storage import PostgresStorage

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Python Data Agent", version="1.0.0")

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # restrict in production
    allow_methods=["*"],
    allow_headers=["*"],
)

# Global configuration
API_KEY = os.getenv("A2A_API_KEY", "dev-key")
storage = PostgresStorage(os.getenv("DATABASE_URL"))
processor = DataProcessor(storage)

# A2A Agent Card
AGENT_CARD = {
    "name": "python-data-processor",
    "version": "1.0.0",
    "description": "Data processing expert supporting CSV/Excel analysis and conversion",
    "url": "http://localhost:8081/a2a",
    "capabilities": {
        "streaming": False,
        "pushNotifications": False
    },
    "skills": [
        {
            "id": "csv-analysis",
            "name": "CSV Data Analysis",
            "description": "Analyze CSV/Excel files, return statistical summaries and visualization suggestions",
            "inputModes": ["text", "file"],
            "outputModes": ["text", "file"]
        },
        {
            "id": "data-cleaning",
            "name": "Data Cleaning",
            "description": "Clean outliers and missing values in datasets",
            "inputModes": ["file"],
            "outputModes": ["file"]
        }
    ],
    "authentication": {
        "type": "apiKey",
        "header": "X-API-Key"
    }
}

# Model definitions
class TaskRequest(BaseModel):
    skill: str
    input: Dict[str, Any]

class TaskResponse(BaseModel):
    task_id: str
    status: str
    result: Optional[Dict[str, Any]] = None
    error: Optional[str] = None

# Authentication
def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key

# Endpoints
@app.get("/.well-known/agent.json")
async def get_agent_card():
    return AGENT_CARD

@app.post("/a2a/tasks")
async def create_task(
    request: TaskRequest,
    api_key: str = Depends(verify_api_key)
):
    import uuid
    task_id = str(uuid.uuid4())

    logger.info(f"Creating task {task_id} for skill {request.skill}")

    try:
        if request.skill == "csv-analysis":
            file_path = request.input.get("file_path")
            analysis_type = request.input.get("analysis_type", "full")

            if not file_path:
                raise ValueError("file_path is required")

            result = processor.process(file_path, analysis_type)

            return TaskResponse(
                task_id=task_id,
                status="completed",
                result={
                    "total_records": result.total_records,
                    "date_range": result.date_range,
                    "revenue_stats": result.revenue_stats,
                    "top_products": result.top_products,
                    "monthly_trend": result.monthly_trend,
                    "anomalies": result.anomalies,
                    "raw_data_path": result.raw_data_path
                }
            )

        elif request.skill == "data-cleaning":
            file_path = request.input.get("file_path")
            if not file_path:
                raise ValueError("file_path is required")

            df = processor._read_data(file_path)
            cleaned_df = processor._clean_data(df)
            output_path = f"/tmp/cleaned_{task_id}.csv"
            cleaned_df.to_csv(output_path, index=False)

            return TaskResponse(
                task_id=task_id,
                status="completed",
                result={"output_path": output_path}
            )

        else:
            raise ValueError(f"Unknown skill: {request.skill}")

    except Exception as e:
        logger.error(f"Task {task_id} failed: {e}")
        return TaskResponse(
            task_id=task_id,
            status="failed",
            error=str(e)
        )

@app.get("/health")
async def health_check():
    return {"status": "healthy", "agent": "python-data-processor"}

if __name__ == "__main__":
    port = int(os.getenv("PORT", "8081"))
    uvicorn.run(app, host="0.0.0.0", port=port)

FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    postgresql-client \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy the code
COPY src/ ./src/

# Non-root user
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8081

CMD ["python", "-m", "src.a2a_server"]

go-gateway-agent/
├── Dockerfile
├── go.mod
├── go.sum
├── main.go
├── internal/
│   ├── gateway/
│   │   └── gateway.go
│   ├── report/
│   │   └── generator.go
│   └── a2a/
│       └── client.go
└── config/
    └── config.yaml

// internal/gateway/gateway.go
package gateway

import (
    "context"
    "fmt"
    "os"
    "time"

    "google.golang.org/adk/a2a/client"
    "google.golang.org/adk/agent"
    "google.golang.org/adk/tool"
)

type GatewayAgent struct {
    agent        *agent.Agent
    pythonClient *a2aclient.Client
    reportClient *a2aclient.Client
}

func NewGatewayAgent(ctx context.Context) (*GatewayAgent, error) {
    // Create the Python Data Agent client
    pythonClient, err := a2aclient.New(ctx,
        a2aclient.WithURL("http://python-agent:8081/a2a"),
        a2aclient.WithAPIKey(os.Getenv("PYTHON_AGENT_API_KEY")),
        a2aclient.WithTimeout(60*time.Second),
        a2aclient.WithRetry(3, time.Second),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to create python client: %w", err)
    }

    // Create the local Report Agent client (internal call)
    reportClient, err := a2aclient.New(ctx,
        a2aclient.WithURL("http://localhost:8082/a2a"),
        a2aclient.WithTimeout(30*time.Second),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to create report client: %w", err)
    }

    // Create the Gateway Agent
    a, err := agent.New(agent.Config{
        Name:        "gateway-agent",
        Model:       model,
        Instruction: `You are the gateway of an intelligent data analysis platform. Your responsibilities: 1. Understand the user's analysis needs 2. Call python-data-processor for data processing 3. Call report-generator to generate the report 4. Aggregate the results and return them to the user`,
        Tools: []tool.Tool{
            NewProcessDataTool(pythonClient),
            NewGenerateReportTool(reportClient),
        },
    })
    if err != nil {
        return nil, err
    }

    return &GatewayAgent{
        agent:        a,
        pythonClient: pythonClient,
        reportClient: reportClient,
    }, nil
}

func (g *GatewayAgent) HandleRequest(ctx context.Context, userInput string) (string, error) {
    return g.agent.Run(ctx, userInput)
}

// internal/report/generator.go
package report

import (
    "context"
    "fmt"
    "os"
    "time"

    "github.com/jung-kurt/gofpdf"
)

type ReportGenerator struct {
    templateDir string
    outputDir   string
}

func NewReportGenerator(templateDir, outputDir string) *ReportGenerator {
    return &ReportGenerator{
        templateDir: templateDir,
        outputDir:   outputDir,
    }
}

func (g *ReportGenerator) GeneratePDF(ctx context.Context, data ReportData) (string, error) {
    pdf := gofpdf.New("P", "mm", "A4", "")
    pdf.AddPage()
    pdf.SetFont("Arial", "B", 16)

    // Title
    pdf.Cell(40, 10, fmt.Sprintf("Data Analysis Report - %s", data.Title))
    pdf.Ln(20)

    // Overview
    pdf.SetFont("Arial", "B", 12)
    pdf.Cell(40, 10, "Data Overview")
    pdf.Ln(10)

    pdf.SetFont("Arial", "", 10)
    pdf.Cell(40, 10, fmt.Sprintf("Total records: %d", data.TotalRecords))
    pdf.Ln(5)
    pdf.Cell(40, 10, fmt.Sprintf("Date range: %s to %s", data.DateRange[0], data.DateRange[1]))
    pdf.Ln(5)
    pdf.Cell(40, 10, fmt.Sprintf("Total revenue: %.2f", data.RevenueStats["total"]))
    pdf.Ln(10)

    // Top selling products
    pdf.SetFont("Arial", "B", 12)
    pdf.Cell(40, 10, "Top Selling Products TOP 10")
    pdf.Ln(10)

    pdf.SetFont("Arial", "", 10)
    for i, product := range data.TopProducts {
        pdf.Cell(40, 10, fmt.Sprintf("%d. %s - Revenue: %.2f",
            i+1, product.Product, product.Revenue))
        pdf.Ln(5)
    }

    // Save
    filename := fmt.Sprintf("report_%d.pdf", time.Now().Unix())
    filepath := fmt.Sprintf("%s/%s", g.outputDir, filename)

    if err := pdf.OutputFileAndClose(filepath); err != nil {
        return "", fmt.Errorf("failed to generate PDF: %w", err)
    }

    return filepath, nil
}

type ReportData struct {
    Title         string
    TotalRecords  int
    DateRange     [2]string
    RevenueStats  map[string]float64
    TopProducts   []ProductStat
    MonthlyTrend  []MonthlyData
}

type ProductStat struct {
    Product  string
    Revenue  float64
    Quantity int
}

type MonthlyData struct {
    Month   string
    Revenue float64
}

// internal/gateway/tools.go
package gateway

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "google.golang.org/adk/a2a/client"
    "google.golang.org/adk/tool"
)

// ProcessDataTool calls the Python Agent to process data
type ProcessDataTool struct {
    client *a2aclient.Client
}

func NewProcessDataTool(client *a2aclient.Client) *ProcessDataTool {
    return &ProcessDataTool{client: client}
}

func (t *ProcessDataTool) Name() string        { return "process_data" }
func (t *ProcessDataTool) Description() string { return "Call the Python Agent to process a data file" }

func (t *ProcessDataTool) Schema() tool.Schema {
    return tool.Schema{
        Type: "object",
        Properties: map[string]tool.Property{
            "file_path": {
                Type:        "string",
                Description: "Path to the data file",
            },
            "analysis_type": {
                Type:        "string",
                Description: "Analysis type: full or summary",
                Enum:        []string{"full", "summary"},
            },
        },
        Required: []string{"file_path"},
    }
}

func (t *ProcessDataTool) Call(ctx context.Context, input string) (string, error) {
    var params struct {
        FilePath     string `json:"file_path"`
        AnalysisType string `json:"analysis_type"`
    }
    if err := json.Unmarshal([]byte(input), &params); err != nil {
        return "", err
    }

    if params.AnalysisType == "" {
        params.AnalysisType = "full"
    }

    // Call the Python Agent
    task, err := t.client.SendTask(ctx, &a2a.Task{
        Input: map[string]interface{}{
            "skill":         "csv-analysis",
            "file_path":     params.FilePath,
            "analysis_type": params.AnalysisType,
        },
    })
    if err != nil {
        return "", fmt.Errorf("failed to send task: %w", err)
    }

    // Wait for the result (simplified; real code should poll)
    result, err := waitForTask(ctx, t.client, task.ID)
    if err != nil {
        return "", err
    }

    return result, nil
}

// GenerateReportTool calls the Report Agent to generate a report
type GenerateReportTool struct {
    client *a2aclient.Client
}

func NewGenerateReportTool(client *a2aclient.Client) *GenerateReportTool {
    return &GenerateReportTool{client: client}
}

func (t *GenerateReportTool) Name() string        { return "generate_report" }
func (t *GenerateReportTool) Description() string { return "Generate a PDF report" }

func (t *GenerateReportTool) Schema() tool.Schema {
    return tool.Schema{
        Type: "object",
        Properties: map[string]tool.Property{
            "data_path": {
                Type:        "string",
                Description: "Path to the data file",
            },
            "title": {
                Type:        "string",
                Description: "Report title",
            },
        },
        Required: []string{"data_path", "title"},
    }
}

func (t *GenerateReportTool) Call(ctx context.Context, input string) (string, error) {
    var params struct {
        DataPath string `json:"data_path"`
        Title    string `json:"title"`
    }
    if err := json.Unmarshal([]byte(input), &params); err != nil {
        return "", err
    }

    task, err := t.client.SendTask(ctx, &a2a.Task{
        Input: map[string]interface{}{
            "skill":      "generate-pdf",
            "data_path":  params.DataPath,
            "title":      params.Title,
        },
    })
    if err != nil {
        return "", fmt.Errorf("failed to send task: %w", err)
    }

    result, err := waitForTask(ctx, t.client, task.ID)
    if err != nil {
        return "", err
    }

    return result, nil
}

func waitForTask(ctx context.Context, client *a2aclient.Client, taskID string) (string, error) {
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    timeout := time.After(5 * time.Minute)

    for {
        select {
        case <-ticker.C:
            task, err := client.GetTask(ctx, taskID)
            if err != nil {
                return "", err
            }

            switch task.Status {
            case a2a.TaskStatusCompleted:
                result, _ := json.Marshal(task.Output)
                return string(result), nil
            case a2a.TaskStatusFailed:
                return "", fmt.Errorf("task failed: %v", task.Output)
            case a2a.TaskStatusCancelled:
                return "", fmt.Errorf("task cancelled")
            }

        case <-timeout:
            client.CancelTask(ctx, taskID)
            return "", fmt.Errorf("task timeout")

        case <-ctx.Done():
            client.CancelTask(ctx, taskID)
            return "", ctx.Err()
        }
    }
}

# docker-compose.yml
version: '3.8'

services:
  python-agent:
    build: ./python-data-agent
    ports:
      - "8081:8081"
    environment:
      - PORT=8081
      - A2A_API_KEY=${PYTHON_AGENT_API_KEY}
      - DATABASE_URL=postgresql://postgres:***@postgres:5432/agentdb
      - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
      - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
    depends_on:
      - postgres
    volumes:
      - ./data:/data
    networks:
      - agent-network

  go-gateway:
    build: ./go-gateway-agent
    ports:
      - "8080:8080"
    environment:
      - PORT=8080
      - PYTHON_AGENT_API_KEY=${PYTHON_AGENT_API_KEY}
      - PYTHON_AGENT_URL=http://python-agent:8081/a2a
      - DATABASE_URL=postgresql://postgres:***@postgres:5432/agentdb
    depends_on:
      - python-agent
      - postgres
    networks:
      - agent-network

  postgres:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=agentdb
    volumes:
      - postgres-data:/var/lib/postgresql/data
    networks:
      - agent-network

  minio:
    image: minio/minio
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      - MINIO_ROOT_USER=minioadmin
      - MINIO_ROOT_PASSWORD=minioadmin
    volumes:
      - minio-data:/data
    networks:
      - agent-network

volumes:
  postgres-data:
  minio-data:

networks:
  agent-network:
    driver: bridge

# 1. Set environment variables
export PYTHON_AGENT_API_KEY="your-secret-key"
export AWS_ACCESS_KEY_ID="your-aws-key"
export AWS_SECRET_ACCESS_KEY="your-aws-secret"

# 2. Start all services
docker-compose up -d

# 3. Verify service status
docker-compose ps
curl http://localhost:8081/health
curl http://localhost:8080/health

# 4. Test the full flow
curl -X POST http://localhost:8080/a2a/tasks \
  -H "Content-Type: application/json" \
  -H "X-API-Key: *** \
  -d '{
    "skill": "analyze-and-report",
    "input": {
      "file_path": "/data/sales_data.csv",
      "title": "Q3 Sales Data Analysis Report"
    }
  }'

// Go side: unified error format
type ErrorResponse struct {
    Code    string `json:"code"`
    Message string `json:"message"`
    Details string `json:"details,omitempty"`
}

func handleA2AError(err error) *ErrorResponse {
    var a2aErr *a2a.Error
    if errors.As(err, &a2aErr) {
        return &ErrorResponse{
            Code:    a2aErr.Code,
            Message: a2aErr.Message,
            Details: a2aErr.Details,
        }
    }

    // Network errors
    if errors.Is(err, context.DeadlineExceeded) {
        return &ErrorResponse{
            Code:    "TIMEOUT",
            Message: "Request timed out, please retry later",
        }
    }

    return &ErrorResponse{
        Code:    "INTERNAL",
        Message: "Internal error",
    }
}

# Python side: unified error format
class AgentError(Exception):
    def __init__(self, code: str, message: str, details: str = None):
        self.code = code
        self.message = message
        self.details = details
        super().__init__(message)

def handle_error(e: Exception) -> dict:
    if isinstance(e, AgentError):
        return {
            "code": e.code,
            "message": e.message,
            "details": e.details
        }
    elif isinstance(e, pd.errors.EmptyDataError):
        return {
            "code": "EMPTY_DATA",
            "message": "The data file is empty"
        }
    elif isinstance(e, ValueError):
        return {
            "code": "INVALID_INPUT",
            "message": str(e)
        }
    else:
        return {
            "code": "INTERNAL",
            "message": "Internal error",
            "details": str(e)
        }