arbitrary_query
Execute custom SQL queries on Logfire database to retrieve and analyze telemetry data, with a lookback period of up to 7 days for precise insights.
Instructions
Run an arbitrary query on the Logfire database.
The schema is available via the `get_logfire_records_schema` tool.
Args:
query: The query to run, as a SQL string.
age: Number of minutes to look back, e.g. 30 for last 30 minutes. Maximum allowed value is 7 days.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| age | Yes | ||
| query | Yes |
Implementation Reference
- logfire_mcp/main.py:62-75 (handler)The main handler function that executes the arbitrary SQL query on the Logfire database, taking a SQL query string and age filter as input.async def arbitrary_query( ctx: Context[ServerSession, MCPState], query: Annotated[str, Field(description='The query to run, as a SQL string.')], age: Age, ) -> list[Any]: """Run an arbitrary query on the Pydantic Logfire database. The SQL reference is available via the `sql_reference` tool. """ logfire_client = ctx.request_context.lifespan_context.logfire_client min_timestamp = datetime.now(UTC) - timedelta(minutes=age) result = await logfire_client.query_json_rows(query, min_timestamp=min_timestamp) return result['rows']
- logfire_mcp/main.py:160-160 (registration)The registration of the arbitrary_query tool using mcp.tool() decorator in the app_factory.mcp.tool()(arbitrary_query)
- logfire_mcp/main.py:24-32 (schema)Type definition and validation schema for the 'age' input parameter used in the arbitrary_query tool and others.Age = Annotated[ int, Field( ge=0, le=7 * DAY, description='Number of minutes to look back, e.g. 30 for last 30 minutes. Maximum allowed value is 7 days.', ), WithJsonSchema({'type': 'integer'}), ]