get_event
Retrieve the source code and details of a Domoticz event script by providing its unique event ID.
Instructions
Get the source code and details of a specific event script by ID.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| event_id | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/domoticz_mcp/server.py:831-835 (handler)The handler function for the 'get_event' tool. It takes an event_id (int), makes an API call to Domoticz to load the event script source code, and returns the response text.
async def get_event(event_id: int) -> str: """Get the source code and details of a specific event script by ID.""" async with create_client() as client: response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=events&evparam=load&event={event_id}") return response.text - src/domoticz_mcp/server.py:830-830 (registration)The tool is registered with the MCP framework using the @mcp.tool() decorator on line 830.
@mcp.tool() - src/domoticz_mcp/server.py:297-298 (helper)The DomoticzClient helper class used by the handler to make authenticated HTTP requests to the Domoticz API.
# Custom AsyncClient wrapper that ensures the token is added class DomoticzClient: - src/domoticz_mcp/server.py:269-296 (helper)The _do_request helper function used by the handler to perform HTTP requests with OAuth retry logic.
async def _do_request(client: httpx.AsyncClient, method: str, url: str, **kwargs) -> httpx.Response: """Perform a request with a single retry on 401 Unauthorized to handle expired tokens.""" global _oauth_token_cache try: resp = await client.request(method, url, **kwargs) if resp.status_code == 401: # Token might be expired. Clear cache and retry once. _oauth_token_cache = None # Re-fetch token (this will trigger OAuth flow if needed) new_token = await _fetch_oauth_token(force_refresh=True) if new_token: # Update headers for the retry if "headers" not in kwargs: kwargs["headers"] = {} kwargs["headers"]["Authorization"] = f"Bearer {new_token}" # Retry the request resp = await client.request(method, url, **kwargs) resp.raise_for_status() return resp except httpx.HTTPStatusError as e: if e.response.status_code == 401: raise Exception("Authentication failed. Please check your credentials or re-authenticate.") raise e