Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/data_models.py:90-134 (handler)

  Complete implementation of the 'get_data_model' MCP tool. Decorated with @mcp_tool for auto-registration using the function name as tool name. Defines input schema via Annotated[str, Field(...)] for 'data_model_id' parameter. The function body fetches the data model details from the Panther REST API endpoint '/data-models/{data_model_id}', handles 404 (not found) and other exceptions, returning structured JSON responses.

  @mcp_tool(
      annotations={
          "permissions": all_perms(Permission.RULE_READ),
          "readOnlyHint": True,
      }
  )
  async def get_data_model(
      data_model_id: Annotated[
          str,
          Field(
              description="The ID of the data model to fetch",
              examples=["MyDataModel", "AWS_CloudTrail", "StandardUser"],
          ),
      ],
  ) -> dict[str, Any]:
      """Get detailed information about a Panther data model, including the mappings and body
  
      Returns complete data model information including Python body code and UDM mappings.
      """
      logger.info(f"Fetching data model details for data model ID: {data_model_id}")
  
      try:
          async with get_rest_client() as client:
              # Allow 404 as a valid response to handle not found case
              result, status = await client.get(
                  f"/data-models/{data_model_id}", expected_codes=[200, 404]
              )
  
              if status == 404:
                  logger.warning(f"No data model found with ID: {data_model_id}")
                  return {
                      "success": False,
                      "message": f"No data model found with ID: {data_model_id}",
                  }
  
          logger.info(
              f"Successfully retrieved data model details for data model ID: {data_model_id}"
          )
          return {"success": True, "data_model": result}
      except Exception as e:
          logger.error(f"Failed to get data model details: {str(e)}")
          return {
              "success": False,
              "message": f"Failed to get data model details: {str(e)}",
          }

• src/mcp_panther/server.py:72-72 (registration)

  Calls register_all_tools(mcp) in the main server setup, which registers all decorated tools including 'get_data_model' with the FastMCP instance.

  register_all_tools(mcp)

• src/mcp_panther/panther_mcp_core/tools/registry.py:75-108 (helper)

  Implementation of register_all_tools that iterates over the global _tool_registry (populated by @mcp_tool decorators), extracts metadata, and invokes mcp_instance.tool(name=func.__name__ or metadata['name'], ...) for each tool.

  def register_all_tools(mcp_instance) -> None:
      """
      Register all tools marked with @mcp_tool with the given MCP instance.
  
      Args:
          mcp_instance: The FastMCP instance to register tools with
      """
      logger.info(f"Registering {len(_tool_registry)} tools with MCP")
  
      # Sort tools by name
      sorted_funcs = sorted(_tool_registry, key=lambda f: f.__name__)
      for tool in sorted_funcs:
          logger.debug(f"Registering tool: {tool.__name__}")
  
          # Get tool metadata if it exists
          metadata = getattr(tool, "_mcp_tool_metadata", {})
  
          annotations = metadata.get("annotations", {})
          # Create tool decorator with metadata
          tool_decorator = mcp_instance.tool(
              name=metadata.get("name"),
              description=metadata.get("description"),
              annotations=annotations,
          )
  
          if annotations and annotations.get("permissions"):
              if not tool.__doc__:
                  tool.__doc__ = ""
              tool.__doc__ += f"\n\n Permissions:{annotations.get('permissions')}"
  
          # Register the tool
          tool_decorator(tool)
  
      logger.info("All tools registered successfully")