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)