Spaces:
Paused
Paused
File size: 2,967 Bytes
026774a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """
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
}
|