Wire Manufacturing Agents to Shop-Floor APIs via Standardised Tool Protocols with MCP SDK and CrewAI

"""
Production implementation for wire manufacturing agents to shop-floor APIs.
This code integrates wire manufacturing agents with shop-floor APIs using standardized tool protocols with MCP SDK and CrewAI.
"""

from typing import Dict, Any, List
import os
import logging
import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.exc import SQLAlchemyError
import time

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

# Database configuration
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///./test.db')
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Data model
class WireData(Base):
    __tablename__ = 'wire_data'
    id = Column(Integer, primary_key=True, index=True)
    wire_type = Column(String, index=True)
    length = Column(Integer)
    diameter = Column(Integer)

# Initialize FastAPI
app = FastAPI()

class Config:
    retries: int = 3  # Number of retries for API calls
    backoff_factor: float = 1.0  # Backoff factor for retries

# Input data model
class WireInput(BaseModel):
    wire_type: str
    length: int
    diameter: int

    @validator('length')
    def check_length(cls, v):
        if v <= 0:
            raise ValueError('Length must be positive')
        return v

    @validator('diameter')
    def check_diameter(cls, v):
        if v <= 0:
            raise ValueError('Diameter must be positive')
        return v

def validate_input(data: WireInput) -> None:
    """Validate request data.
    
    Args:
        data: Input to validate
    Raises:
        ValueError: If validation fails
    """
    logger.info('Validating input data...')
    if not all([data.wire_type, data.length, data.diameter]):
        raise ValueError('All fields are required')

def save_to_db(data: WireInput) -> None:
    """Save wire data to the database.
    
    Args:
        data: WireInput object to save
    Raises:
        SQLAlchemyError: If database operation fails
    """
    logger.info('Saving data to the database...')
    db = SessionLocal()  # Create a new database session
    wire_entry = WireData(wire_type=data.wire_type, length=data.length, diameter=data.diameter)
    try:
        db.add(wire_entry)
        db.commit()
        logger.info('Data saved successfully')
    except SQLAlchemyError as e:
        logger.error(f'Database error: {e}')
        db.rollback()
        raise e
    finally:
        db.close()

def call_api(data: WireInput) -> Dict[str, Any]:
    """Call external shop-floor API.
    
    Args:
        data: WireInput object to send
    Returns:
        Response from the API
    Raises:
        HTTPException: If API call fails
    """
    logger.info('Calling external API...')
    url = 'https://api.shopfloor.com/wire'
    for attempt in range(Config.retries):
        try:
            response = requests.post(url, json=data.dict())
            response.raise_for_status()  # Raise error for bad responses
            logger.info('API call successful')
            return response.json()
        except requests.HTTPError as e:
            logger.error(f'API call failed: {e}')
            time.sleep(Config.backoff_factor * (2 ** attempt))
    raise HTTPException(status_code=500, detail='Failed to reach external API')

@app.post('/wire', response_model=Dict[str, Any])
async def create_wire(data: WireInput):
    """Create wire entry and call shop-floor API.
    
    Args:
        data: WireInput object to process
    Returns:
        API response
    Raises:
        HTTPException: If an error occurs
    """
    validate_input(data)  # Validate incoming data
    save_to_db(data)  # Save data to the database
    api_response = call_api(data)  # Call external API
    return api_response

if __name__ == '__main__':
    # Create database tables
    Base.metadata.create_all(bind=engine)
    logger.info('Database tables created')
    pass