Build A2A-Compatible Supply Chain Agents with CrewAI and FastAPI

"""
Production implementation for A2A-Compatible Supply Chain Agents.
Provides secure, scalable operations using CrewAI and FastAPI.
"""

from typing import Dict, Any, List, Tuple
import os
import logging
import httpx
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, constr
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# Logger setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class for environment variables
class Config:
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///./test.db')
    api_base_url: str = os.getenv('API_BASE_URL', 'http://api.example.com')

# SQLAlchemy setup
Base = declarative_base()
engine = create_engine(Config.database_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Data model for the supply chain agent
class SupplyChainAgent(Base):
    __tablename__ = 'supply_chain_agents'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    status = Column(String)

Base.metadata.create_all(bind=engine)  # Create tables if not exist

# FastAPI application
app = FastAPI()

# Helper functions
async def validate_input(data: Dict[str, Any]) -> None:
    """Validate request data.
    
    Args:
        data: Input to validate
    Raises:
        ValueError: If validation fails
    """
    if 'name' not in data:
        raise ValueError('Missing name in input data.')
    if not isinstance(data['name'], str):
        raise ValueError('Name must be a string.')

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields.
    
    Args:
        data: Input data
    Returns:
        Sanitized data
    """
    return {'name': data['name'].strip()}

async def fetch_data(agent_id: int) -> SupplyChainAgent:
    """Fetch agent data from the database.
    
    Args:
        agent_id: ID of the agent
    Returns:
        SupplyChainAgent: The fetched agent
    Raises:
        HTTPException: If agent not found
    """
    with SessionLocal() as session:
        agent = session.query(SupplyChainAgent).filter(SupplyChainAgent.id == agent_id).first()
        if not agent:
            raise HTTPException(status_code=404, detail='Agent not found')
        return agent

async def save_to_db(agent_data: Dict[str, Any]) -> SupplyChainAgent:
    """Save agent data to the database.
    
    Args:
        agent_data: Data to save
    Returns:
        SupplyChainAgent: The saved agent
    """
    with SessionLocal() as session:
        agent = SupplyChainAgent(**agent_data)
        session.add(agent)
        session.commit()
        session.refresh(agent)
        return agent

async def call_api(endpoint: str, params: Dict[str, Any]) -> Any:
    """Call an external API.
    
    Args:
        endpoint: API endpoint
        params: Parameters for the request
    Returns:
        JSON response from the API
    Raises:
        HTTPException: If API call fails
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(f'{Config.api_base_url}/{endpoint}', params=params)
        if response.status_code != 200:
            raise HTTPException(status_code=response.status_code, detail='API call failed')
        return response.json()

# Main class orchestrating operations
class SupplyChainManager:
    async def create_agent(self, data: Dict[str, Any]) -> SupplyChainAgent:
        """Create a new supply chain agent.
        
        Args:
            data: Input data for the agent
        Returns:
            SupplyChainAgent: The created agent
        """
        await validate_input(data)
        sanitized_data = await sanitize_fields(data)
        return await save_to_db(sanitized_data)

# API endpoints
@app.post('/agents/', response_model=SupplyChainAgent)
async def create_agent(request: Request):
    """Create a new supply chain agent.
    
    Args:
        request: Incoming request
    Returns:
        SupplyChainAgent: The created agent
    """
    data = await request.json()
    manager = SupplyChainManager()
    return await manager.create_agent(data)

@app.get('/agents/{agent_id}', response_model=SupplyChainAgent)
async def read_agent(agent_id: int):
    """Read supply chain agent by ID.
    
    Args:
        agent_id: ID of the agent
    Returns:
        SupplyChainAgent: The fetched agent
    """
    return await fetch_data(agent_id)

if __name__ == '__main__':
    # Example usage
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)  # Start the FastAPI server