arbitrary_query
Execute custom SQL queries on Logfire telemetry data to analyze traces, metrics, and exceptions within specified time windows.
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 7 days. |
Implementation Reference
- logfire_mcp/main.py:62-75 (handler)The handler function that executes an arbitrary SQL query on the Logfire database, using the provided query string and age (minutes to look back).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:150-164 (registration)The app_factory function creates the FastMCP server instance and registers the arbitrary_query tool along with others using mcp.tool()def app_factory(logfire_read_token: str, logfire_base_url: str | None = None) -> FastMCP: @asynccontextmanager async def lifespan(server: FastMCP) -> AsyncIterator[MCPState]: # print to stderr so we this message doesn't get read by the MCP client headers = {'User-Agent': f'logfire-mcp/{__version__}'} async with AsyncLogfireQueryClient(logfire_read_token, headers=headers, base_url=logfire_base_url) as client: yield MCPState(logfire_client=client) mcp = FastMCP('Logfire', lifespan=lifespan) mcp.tool()(find_exceptions_in_file) mcp.tool()(arbitrary_query) mcp.tool()(logfire_link) mcp.tool()(schema_reference) return mcp
- logfire_mcp/main.py:24-32 (schema)Type definition for the 'age' parameter used in arbitrary_query, validating it as an integer between 0 and 7 days (10080 minutes).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'}), ]