get_scheduled_query
Retrieve detailed information about a specific scheduled query by ID, including SQL code, schedule configuration, and metadata for security monitoring analysis.
Instructions
Get detailed information about a specific scheduled query by ID.
Returns complete scheduled query information including SQL, schedule configuration, and metadata.
Returns: Dict containing: - success: Boolean indicating if the query was successful - query: Scheduled query information if successful, containing: - id: Query ID - name: Query name - description: Query description - sql: The SQL query text - schedule: Schedule configuration (cron, rate, timeout) - managed: Whether the query is managed by Panther - createdAt: Creation timestamp - updatedAt: Last update timestamp - message: Error message if unsuccessful
Permissions:{'all_of': ['Query Data Lake']}
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query_id | Yes | The ID of the scheduled query to fetch (must be a UUID) |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The core handler function that executes the get_scheduled_query tool. It takes a UUID query_id, fetches the query details from the Panther REST API endpoint /queries/{query_id}, and returns a structured response with success status and query data or error message.
async def get_scheduled_query( query_id: Annotated[ UUID, Field( description="The ID of the scheduled query to fetch (must be a UUID)", examples=["6c6574cb-fbf9-49fc-baad-1a99464ef09e"], ), ], ) -> Dict[str, Any]: """Get detailed information about a specific scheduled query by ID. Returns complete scheduled query information including SQL, schedule configuration, and metadata. Returns: Dict containing: - success: Boolean indicating if the query was successful - query: Scheduled query information if successful, containing: - id: Query ID - name: Query name - description: Query description - sql: The SQL query text - schedule: Schedule configuration (cron, rate, timeout) - managed: Whether the query is managed by Panther - createdAt: Creation timestamp - updatedAt: Last update timestamp - message: Error message if unsuccessful """ logger.info(f"Fetching scheduled query: {query_id}") try: # Execute the REST API call async with get_rest_client() as client: response_data, status_code = await client.get(f"/queries/{str(query_id)}") logger.info(f"Successfully retrieved scheduled query: {query_id}") # Format the response return { "success": True, "query": response_data, } except Exception as e: logger.error(f"Failed to fetch scheduled query: {str(e)}") return { "success": False, "message": f"Failed to fetch scheduled query: {str(e)}", } - Input schema definition for the tool using Pydantic's Annotated with UUID type and Field providing description and example.
query_id: Annotated[ UUID, Field( description="The ID of the scheduled query to fetch (must be a UUID)", examples=["6c6574cb-fbf9-49fc-baad-1a99464ef09e"], ), ], ) -> Dict[str, Any]: - The @mcp_tool decorator registers the get_scheduled_query function into the global tool registry with required permissions and read-only annotation.
@mcp_tool( annotations={ "permissions": all_perms(Permission.DATA_ANALYTICS_READ), "readOnlyHint": True, } ) - src/mcp_panther/server.py:71-72 (registration)Invocation of register_all_tools on the FastMCP instance, which iterates over all decorated tools (including get_scheduled_query) and registers them with the MCP server using mcp_instance.tool() with extracted metadata.
# Register all tools with MCP using the registry register_all_tools(mcp)