"""
Laravel API Integration Layer
This module provides a unified interface for communicating with the Laravel backend.
It handles authentication, error handling, and response parsing for all API calls.
Security: All requests include proper Bearer token authentication
"""
from typing import Dict, Any, Optional
import httpx
from urllib.parse import urljoin
from src.config import settings
from src.observability import get_logger
logger = get_logger(__name__)
class LaravelAPI:
"""
Laravel API client for making authenticated requests to the backend.
This class handles HTTP requests to the Laravel API with proper authentication
and error handling. It's used by all MCP tools to communicate with the backend.
"""
def __init__(self, bearer_token: str):
"""
Initialize the API client with authentication token.
Args:
bearer_token: Laravel Sanctum token for authentication
"""
self.bearer_token = bearer_token
self.base_url = settings.laravel_api_url
self.timeout = 30.0 # 30 second timeout
logger.debug(
"laravel_api_initialized",
base_url=self.base_url,
timeout=self.timeout
)
async def get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Make a GET request to the Laravel API.
Args:
endpoint: API endpoint (e.g., "/api/v2/list-vendors")
params: Query parameters
Returns:
Parsed JSON response from the API
Raises:
httpx.HTTPError: For HTTP-related errors
ValueError: For invalid JSON responses
"""
return await self._request("GET", endpoint, params=params)
async def post(self, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Make a POST request to the Laravel API.
Args:
endpoint: API endpoint (e.g., "/api/v2/create-vendor")
data: Request body data
Returns:
Parsed JSON response from the API
Raises:
httpx.HTTPError: For HTTP-related errors
ValueError: For invalid JSON responses
"""
return await self._request("POST", endpoint, json=data)
async def put(self, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Make a PUT request to the Laravel API.
Args:
endpoint: API endpoint (e.g., "/api/v2/update-vendor/123")
data: Request body data
Returns:
Parsed JSON response from the API
Raises:
httpx.HTTPError: For HTTP-related errors
ValueError: For invalid JSON responses
"""
return await self._request("PUT", endpoint, json=data)
async def delete(self, endpoint: str) -> Dict[str, Any]:
"""
Make a DELETE request to the Laravel API.
Args:
endpoint: API endpoint (e.g., "/api/v2/delete-vendor/123")
Returns:
Parsed JSON response from the API
Raises:
httpx.HTTPError: For HTTP-related errors
ValueError: For invalid JSON responses
"""
return await self._request("DELETE", endpoint)
async def _request(
self,
method: str,
endpoint: str,
params: Optional[Dict[str, Any]] = None,
json: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Internal method to make HTTP requests with proper error handling.
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint
params: Query parameters (for GET requests)
json: Request body (for POST/PUT requests)
Returns:
Parsed JSON response
Raises:
httpx.HTTPError: For HTTP-related errors
ValueError: For invalid JSON responses
"""
# Build full URL
url = urljoin(self.base_url, endpoint)
# Prepare headers
headers = {
"Authorization": f"Bearer {self.bearer_token}",
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": f"Updation-MCP/{settings.mcp_server_name}"
}
logger.debug(
"laravel_api_request_started",
method=method,
url=url,
has_params=bool(params),
has_json=bool(json)
)
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.request(
method=method,
url=url,
headers=headers,
params=params,
json=json
)
# Log response details
logger.debug(
"laravel_api_response_received",
method=method,
url=url,
status_code=response.status_code,
response_size=len(response.content)
)
# Handle HTTP errors
if response.status_code >= 400:
error_data = {}
try:
error_data = response.json()
except ValueError:
error_data = {"raw_response": response.text}
logger.error(
"laravel_api_http_error",
method=method,
url=url,
status_code=response.status_code,
error_data=error_data
)
# Return error response in consistent format
return {
"success": False,
"message": error_data.get("message", f"HTTP {response.status_code} error"),
"error": error_data.get("error", response.text),
"status_code": response.status_code
}
# Parse successful response
try:
data = response.json()
logger.debug(
"laravel_api_request_success",
method=method,
url=url,
response_success=data.get("success", True)
)
return data
except ValueError as e:
logger.error(
"laravel_api_json_parse_error",
method=method,
url=url,
error=str(e),
response_text=response.text[:500] # First 500 chars
)
return {
"success": False,
"message": "Invalid JSON response from server",
"error": str(e),
"raw_response": response.text
}
except httpx.TimeoutException as e:
logger.error(
"laravel_api_timeout_error",
method=method,
url=url,
timeout=self.timeout,
error=str(e)
)
return {
"success": False,
"message": "Request timeout",
"error": f"Request timed out after {self.timeout} seconds"
}
except httpx.ConnectError as e:
logger.error(
"laravel_api_connection_error",
method=method,
url=url,
error=str(e)
)
return {
"success": False,
"message": "Connection error",
"error": "Unable to connect to Laravel API"
}
except httpx.HTTPError as e:
logger.error(
"laravel_api_http_error",
method=method,
url=url,
error=str(e)
)
return {
"success": False,
"message": "HTTP request failed",
"error": str(e)
}
except Exception as e:
logger.error(
"laravel_api_unexpected_error",
method=method,
url=url,
error=str(e),
error_type=type(e).__name__
)
return {
"success": False,
"message": "Unexpected error occurred",
"error": str(e)
}
class LaravelAPIError(Exception):
"""
Custom exception for Laravel API errors.
This exception is raised when the API returns an error response
that cannot be handled gracefully by the calling code.
"""
def __init__(self, message: str, error_data: Optional[Dict[str, Any]] = None):
super().__init__(message)
self.message = message
self.error_data = error_data or {}
def __str__(self):
return f"LaravelAPIError: {self.message}"
