MWAA MCP Server

Instructions

Implementation Reference

• awslabs/mwaa_mcp_server/tools.py:462-472 (handler)

  The implementation of the get_import_errors tool logic which calls the Airflow API.

  async def get_import_errors(
      self,
      environment_name: str,
      limit: Optional[int] = 100,
      offset: Optional[int] = 0,
  ) -> Dict[str, Any]:
      """Get import errors via Airflow API."""
      params: Dict[str, Any] = {"limit": limit, "offset": offset}
      return self._invoke_airflow_api(
          environment_name, "GET", "/dags/importErrors", params=params
      )

• awslabs/mwaa_mcp_server/server.py:557-576 (registration)

  Registration and handler wrapper for the get_import_errors tool in the MCP server.

  @mcp.tool(name="get_import_errors")
  async def get_import_errors(
      environment_name: str,
      limit: Optional[int] = 100,
      offset: Optional[int] = 0,
  ) -> Dict[str, Any]:
      """Get DAG import errors 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 import errors
      """
      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.get_import_errors(environment_name, limit_int, offset_int)