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

Class BaseAPIHandler

api/api_handlers.py:59–94  ·  view source on GitHub ↗

Base class for asynchronous API handlers.

Source from the content-addressed store, hash-verified

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

Callers

nothing calls this directly

Calls

no outgoing calls

Tested by

no test coverage detected