MWAA MCP Server

Instructions

Implementation Reference

• awslabs/mwaa_mcp_server/tools.py:450-460 (handler)

  The actual implementation of the list_variables tool logic that interacts with the Airflow API.

  async def list_variables(
      self,
      environment_name: str,
      limit: Optional[int] = 100,
      offset: Optional[int] = 0,
  ) -> Dict[str, Any]:
      """List variables via Airflow API."""
      params: Dict[str, Any] = {"limit": limit, "offset": offset}
      return self._invoke_airflow_api(
          environment_name, "GET", "/variables", params=params
      )

• awslabs/mwaa_mcp_server/server.py:535-554 (registration)

  The MCP tool registration and entry point wrapper for list_variables.

  @mcp.tool(name="list_variables")
  async def list_variables(
      environment_name: str,
      limit: Optional[int] = 100,
      offset: Optional[int] = 0,
  ) -> Dict[str, Any]:
      """List all Airflow variables in the environment.
  
      Args:
          environment_name: Name of the MWAA environment
          limit: Number of items to return
          offset: Number of items to skip
  
      Returns:
          Dictionary containing list of variables
      """
      limit_int = int(limit) if limit is not None else 100
      offset_int = int(offset) if offset is not None else 0
  
      return await tools.list_variables(environment_name, limit_int, offset_int)