Spaces:
Paused
Paused
| """ | |
| Plumber API Client for TSA Agent Tools | |
| This module provides async HTTP client functionality for communicating | |
| with the R Plumber API that bridges to the Java TSA engine. | |
| """ | |
| import os | |
| import httpx | |
| from typing import Any | |
| def get_api_url() -> str: | |
| """Get the TSA API base URL from environment.""" | |
| return os.environ.get("TSA_API_URL", "http://localhost:8000") | |
| async def api_call( | |
| method: str, | |
| endpoint: str, | |
| params: dict[str, Any] | None = None, | |
| json_data: dict[str, Any] | None = None, | |
| timeout: float = 30.0, | |
| ) -> dict[str, Any]: | |
| """ | |
| Make an async API call to the Plumber API. | |
| Args: | |
| method: HTTP method (GET, POST, DELETE) | |
| endpoint: API endpoint (e.g., "/analysis/create") | |
| params: Query parameters | |
| json_data: JSON body data | |
| timeout: Request timeout in seconds | |
| Returns: | |
| API response as dictionary | |
| """ | |
| base_url = get_api_url() | |
| url = f"{base_url}{endpoint}" | |
| async with httpx.AsyncClient(timeout=timeout) as client: | |
| try: | |
| if method.upper() == "GET": | |
| response = await client.get(url, params=params) | |
| elif method.upper() == "POST": | |
| response = await client.post(url, params=params, json=json_data) | |
| elif method.upper() == "DELETE": | |
| response = await client.delete(url, params=params) | |
| else: | |
| raise ValueError(f"Unsupported HTTP method: {method}") | |
| response.raise_for_status() | |
| return response.json() | |
| except httpx.ConnectError: | |
| return { | |
| "success": False, | |
| "message": f"Cannot connect to TSA API at {base_url}. " | |
| "Ensure the Plumber server is running." | |
| } | |
| except httpx.HTTPStatusError as e: | |
| return { | |
| "success": False, | |
| "message": f"API error: {e.response.status_code} - {e.response.text}" | |
| } | |
| except Exception as e: | |
| return { | |
| "success": False, | |
| "message": f"Unexpected error: {str(e)}" | |
| } | |
| def format_tool_response( | |
| data: dict[str, Any] | str, | |
| is_error: bool = False | |
| ) -> dict[str, Any]: | |
| """ | |
| Format a response for the MCP tool protocol. | |
| Args: | |
| data: Response data (dict or string) | |
| is_error: Whether this is an error response | |
| Returns: | |
| Properly formatted MCP tool response | |
| """ | |
| if isinstance(data, str): | |
| text = data | |
| elif isinstance(data, dict): | |
| if data.get("success") is False: | |
| is_error = True | |
| text = data.get("message", "Unknown error") | |
| else: | |
| # Format the dict nicely | |
| import json | |
| text = json.dumps(data, indent=2, default=str) | |
| else: | |
| text = str(data) | |
| return { | |
| "content": [{"type": "text", "text": text}], | |
| "is_error": is_error | |
| } | |