Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/scheduled_queries.py:21-116 (handler)

  Handler function for list_scheduled_queries tool. Includes @mcp_tool decorator for registration, inline Pydantic schema via Annotated[Field], fetches scheduled queries via REST API, removes SQL to avoid token limits, supports pagination and name filtering, returns formatted success/error response.
  @mcp_tool( annotations={ "permissions": all_perms(Permission.DATA_ANALYTICS_READ), "readOnlyHint": True, } ) async def list_scheduled_queries( cursor: Annotated[ str | None, Field(description="Optional cursor for pagination from a previous query"), ] = None, limit: Annotated[ int, Field( description="Maximum number of results to return (1-1000)", ge=1, le=1000, ), ] = 100, name_contains: Annotated[ str | None, Field( description="Optional substring to filter scheduled queries by name (case-insensitive)" ), ] = None, ) -> Dict[str, Any]: """List all scheduled queries from your Panther instance. Scheduled queries are SQL queries that run automatically on a defined schedule for recurring analysis, reporting, and monitoring tasks. Note: SQL content is excluded from list responses to prevent token limits. Use get_scheduled_query() to retrieve the full SQL for a specific query. Returns: Dict containing: - success: Boolean indicating if the query was successful - queries: List of scheduled queries if successful, each containing: - id: Query ID - name: Query name - description: Query description - schedule: Schedule configuration (cron, rate, timeout) - managed: Whether the query is managed by Panther - createdAt: Creation timestamp - updatedAt: Last update timestamp - total_queries: Number of queries returned - has_next_page: Boolean indicating if more results are available - next_cursor: Cursor for fetching the next page of results - message: Error message if unsuccessful """ logger.info("Listing scheduled queries") try: # Prepare query parameters params = {"limit": limit} if cursor: params["cursor"] = cursor logger.debug(f"Query parameters: {params}") # Execute the REST API call async with get_rest_client() as client: response_data, status_code = await client.get("/queries", params=params) # Extract queries from response queries = response_data.get("results", []) next_cursor = response_data.get("next") # Remove SQL content to prevent token limit issues # Full SQL can be retrieved using get_scheduled_query for query in queries: if "sql" in query: del query["sql"] # Filter by name_contains if provided if name_contains: queries = [ q for q in queries if name_contains.lower() in q.get("name", "").lower() ] logger.info(f"Successfully retrieved {len(queries)} scheduled queries") # Format the response return { "success": True, "queries": queries, "total_queries": len(queries), "has_next_page": bool(next_cursor), "next_cursor": next_cursor, } except Exception as e: logger.error(f"Failed to list scheduled queries: {str(e)}") return { "success": False, "message": f"Failed to list scheduled queries: {str(e)}", }
• src/mcp_panther/panther_mcp_core/tools/__init__.py:25-38 (registration)

  Import of scheduled_queries module in tools/__init__.py triggers execution of @mcp_tool decorators, adding the tool to the internal registry.
  from . import ( alerts, data_lake, data_models, detections, global_helpers, metrics, permissions, roles, scheduled_queries, schemas, sources, users, )
• src/mcp_panther/server.py:65-72 (registration)

  Final registration of all tools (including list_scheduled_queries) with the MCP server instance by calling register_all_tools(mcp). Note: exact lines approximate; full context confirms the call.
  from .panther_mcp_core.tools.registry import register_all_tools # Create the MCP server with lifespan context for shared HTTP client management # Note: Dependencies are declared in fastmcp.json for FastMCP v2.14.0+ mcp = FastMCP(MCP_SERVER_NAME, lifespan=lifespan) # Register all tools with MCP using the registry register_all_tools(mcp)