"""
Pattern 94: API Integration Pattern

Agents that integrate with external APIs (REST, GraphQL, etc.) to fetch data,
trigger actions, or interact with external systems.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import json


class HTTPMethod(Enum):
    """HTTP methods."""
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"


@dataclass
class APIEndpoint:
    """API endpoint configuration."""
    name: str
    base_url: str
    path: str
    method: HTTPMethod
    headers: Dict[str, str] = field(default_factory=dict)
    auth_required: bool = False
    description: str = ""


@dataclass
class APICall:
    """Track API calls."""
    endpoint: str
    method: str
    params: Dict[str, Any]
    response: Any
    status_code: int
    timestamp: datetime = field(default_factory=datetime.now)
    execution_time: float = 0.0


class APIIntegrationPattern:
    """
    API Integration Pattern - Agent with API integration capabilities.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.endpoints: Dict[str, APIEndpoint] = {}
        self.call_history: List[APICall] = []
    
    def register_endpoint(self, endpoint: APIEndpoint):
        """Register an API endpoint."""
        self.endpoints[endpoint.name] = endpoint
    
    def call_api(self, endpoint_name: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """
        Simulate API call (in real implementation, would use requests library).
        """
        if endpoint_name not in self.endpoints:
            return {"error": f"Endpoint {endpoint_name} not found"}
        
        endpoint = self.endpoints[endpoint_name]
        params = params or {}
        
        # Simulate API response
        start = datetime.now()
        
        # Mock response based on endpoint
        mock_response = self._generate_mock_response(endpoint, params)
        
        execution_time = (datetime.now() - start).total_seconds()
        
        # Track call
        call = APICall(
            endpoint=endpoint_name,
            method=endpoint.method.value,
            params=params,
            response=mock_response,
            status_code=200,
            execution_time=execution_time
        )
        self.call_history.append(call)
        
        return {
            "endpoint": endpoint_name,
            "status": "success",
            "data": mock_response,
            "execution_time": execution_time
        }
    
    def _generate_mock_response(self, endpoint: APIEndpoint, params: Dict[str, Any]) -> Any:
        """Generate mock API response."""
        # Simulate different responses based on endpoint
        if "user" in endpoint.name.lower():
            return {"id": 1, "name": "John Doe", "email": "john@example.com"}
        elif "weather" in endpoint.name.lower():
            return {"temperature": 72, "condition": "sunny", "humidity": 45}
        elif "data" in endpoint.name.lower():
            return {"items": [{"id": i, "value": f"item_{i}"} for i in range(5)]}
        else:
            return {"message": "Success", "data": params}
    
    def run(self, task: str) -> Dict[str, Any]:
        """Execute task using API integration."""
        # Use LLM to decide which API to call
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You have access to these APIs: {list(self.endpoints.keys())}. Decide which to call based on the task."),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        
        # Simple parsing (in production, would use function calling)
        response_text = str(response.content)
        
        # Call APIs mentioned in response
        results = []
        for endpoint_name in self.endpoints.keys():
            if endpoint_name.lower() in response_text.lower():
                result = self.call_api(endpoint_name)
                results.append(result)
        
        return {
            "task": task,
            "endpoints_called": len(results),
            "results": results,
            "analysis": response.content
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get API call statistics."""
        return {
            "total_calls": len(self.call_history),
            "endpoints_registered": len(self.endpoints),
            "calls_by_endpoint": {
                name: sum(1 for call in self.call_history if call.endpoint == name)
                for name in self.endpoints.keys()
            }
        }


# Example usage
if __name__ == "__main__":
    print("=== API Integration Pattern Demo ===\n")
    
    pattern = APIIntegrationPattern()
    
    # Register endpoints
    pattern.register_endpoint(APIEndpoint(
        name="GetUser",
        base_url="https://api.example.com",
        path="/users/{id}",
        method=HTTPMethod.GET,
        description="Fetch user information"
    ))
    
    pattern.register_endpoint(APIEndpoint(
        name="GetWeather",
        base_url="https://api.weather.com",
        path="/current",
        method=HTTPMethod.GET,
        description="Get current weather"
    ))
    
    pattern.register_endpoint(APIEndpoint(
        name="GetData",
        base_url="https://api.data.com",
        path="/items",
        method=HTTPMethod.GET,
        description="Fetch data items"
    ))
    
    # Test task
    result = pattern.run("Get user information and current weather")
    print(f"Task: {result['task']}")
    print(f"Endpoints called: {result['endpoints_called']}")
    print(f"\nResults:")
    for r in result['results']:
        print(f"  - {r['endpoint']}: {r['data']}")
    
    # Statistics
    print(f"\nStatistics: {pattern.get_statistics()}")

    

Agentic Architecture Pattern: Api Integration

Pattern 94: 94 Api Integration – Line-by-Line Explanation

Overview

Pattern 94: API Integration Pattern

Agents that integrate with external APIs (REST, GraphQL, etc.) to fetch data, trigger actions, or interact with external systems.

Code Structure Analysis

File Information

• Pattern Number: 94
• Pattern Name: 94 Api Integration
• File Path: patterns\tool_resource\94_api_integration.py
• Total Lines: 193

Detailed Code Explanation

Module Docstring

"""
Pattern 94: API Integration Pattern

Agents that integrate with external APIs (REST, GraphQL, etc.) to fetch data,
trigger actions, or interact with external systems.
"""

Explanation: This module docstring describes the purpose and key features of the 94 Api Integration pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import json  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – import json: JSON parsing and serialization

Enumerations

HTTPMethod (Line 17)

class HTTPMethod(Enum):
    GET
    POST
    PUT
    DELETE
    PATCH

Purpose: HTTP methods.

Values: – GET: Represents get – POST: Represents post – PUT: Represents put – DELETE: Represents delete – PATCH: Represents patch

Classes

HTTPMethod (Line 17)

Description: HTTP methods.

APIEndpoint (Line 27)

Description: API endpoint configuration.

APICall (Line 39)

Description: Track API calls.

APIIntegrationPattern (Line 50)

Description: API Integration Pattern – Agent with API integration capabilities.

Methods:

__init__ (Line 55)

def __init__(self, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”)

register_endpoint (Line 60)

def register_endpoint(self, endpoint):
    """Register an API endpoint."""

Purpose: Register an API endpoint.

Parameters: – self: Instance reference – endpoint: Parameter for endpoint

call_api (Line 64)

def call_api(self, endpoint_name, params):
    """Simulate API call (in real implementation, would use requests library)."""

Purpose: Simulate API call (in real implementation, would use requests library).

Parameters: – self: Instance reference – endpoint_name: Parameter for endpoint_name – params: Parameter for params

_generate_mock_response (Line 100)

def _generate_mock_response(self, endpoint, params):
    """Generate mock API response."""

Purpose: Generate mock API response.

Parameters: – self: Instance reference – endpoint: Parameter for endpoint – params: Parameter for params

run (Line 112)

def run(self, task):
    """Execute task using API integration."""

Purpose: Execute task using API integration.

Parameters: – self: Instance reference – task: Parameter for task

get_statistics (Line 140)

def get_statistics(self):
    """Get API call statistics."""

Purpose: Get API call statistics.

Parameters: – self: Instance reference

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = HTTPMethod(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities