arbitrary_query
Run custom SQL queries to analyze telemetry data from Logfire, with a configurable time window.
Instructions
Run an arbitrary query on the Pydantic Logfire database.
The SQL reference is available via the `sql_reference` tool.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The query to run, as a SQL string. | |
| age | Yes | Number of minutes to look back, e.g. 30 for last 30 minutes. Maximum allowed value is 30 days. |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- logfire_mcp/main.py:62-74 (handler)The handler function that executes the 'arbitrary_query' tool logic. It takes a SQL query string and age parameter, queries the Logfire database, and returns the results.
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 `schema_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:24-32 (schema)The 'Age' type alias used as a schema for the 'age' parameter in arbitrary_query. It's an integer constrained between 0 and 30 days (in minutes).
Age = Annotated[ int, Field( ge=0, le=30 * DAY, description='Number of minutes to look back, e.g. 30 for last 30 minutes. Maximum allowed value is 30 days.', ), WithJsonSchema({'type': 'integer'}), ] - logfire_mcp/main.py:64-66 (schema)Input parameter schema for arbitrary_query: query (a SQL string) and age (constrained integer minutes).
query: Annotated[str, Field(description='The query to run, as a SQL string.')], age: Age, ) -> list[Any]: - logfire_mcp/main.py:160-160 (registration)Registration of the arbitrary_query function as an MCP tool via the FastMCP framework.
mcp.tool()(arbitrary_query)