MCPcopy Create free account
hub / github.com/Roy3838/Observer / BaseAPIHandler

Class BaseAPIHandler

api/api_handlers.py:60–95  ·  view source on GitHub ↗

Base class for asynchronous API handlers.

Source from the content-addressed store, hash-verified

58
59
60class BaseAPIHandler:
61 """Base class for asynchronous API handlers."""
62 def __init__(self, name):
63 self.name = name
64 self.models = [] # List of supported models { "name": "model-id", "parameters": "optional", ... }
65 API_HANDLERS[name] = self
66 logger.info("Registered API handler: '%s'", name)
67 # Optional: Create a shared httpx client if needed across handlers (managing lifecycle is key)
68 # self.http_client = httpx.AsyncClient(timeout=90.0) # Example
69
70 def get_models(self):
71 """Return the list of models supported by this handler."""
72 return self.models
73
74 async def handle_request(self, request_data: dict) -> dict:
75 """
76 Process the request asynchronously.
77 Subclasses MUST override this method.
78
79 Args:
80 request_data: The parsed JSON request data (dictionary).
81
82 Returns:
83 A dictionary representing the successful JSON response payload.
84
85 Raises:
86 ConfigError: If configuration (like API key) is missing.
87 BackendAPIError: If the downstream API call fails.
88 ValueError: If the request_data is invalid.
89 NotImplementedError: If the subclass doesn't implement this.
90 Exception: For other unexpected errors.
91 """
92 raise NotImplementedError(f"Handler '{self.name}' must implement handle_request")
93
94 # NOTE: log_conversation() method removed - now handled centrally in compute.py
95 # This eliminates disk I/O during requests and ensures consistent logging across all handlers
96
97
98# --- Import and Instantiate REWRITTEN Handlers ---

Callers

nothing calls this directly

Calls

no outgoing calls

Tested by

no test coverage detected