list_environments

Retrieve all Amazon MWAA environments in your AWS account and region to manage Apache Airflow workflows and monitor operations.

Instructions

List all MWAA environments in the current AWS account and region.

Args: max_results: Maximum number of environments to return (1-25)

Returns: Dictionary containing list of environment names and metadata

Input Schema

TableJSON Schema

Name	Required	Description	Default
`max_results`	No

Implementation Reference

awslabs/mwaa_mcp_server/tools.py:76-114 (handler)

The handler method in `MWAATools` class that interacts with Boto3 to list MWAA environments and get their details.

async def list_environments(self, max_results: Optional[int] = None) -> Dict[str, Any]:
    """List MWAA environments."""
    try:
        kwargs: Dict[str, Any] = {}
        if max_results:
            kwargs["MaxResults"] = min(max_results, 25)

        response = self.mwaa_client.list_environments(**kwargs)

        environments = []
        for env_name in response.get("Environments", []):
            try:
                env_details = await self.get_environment(env_name)
                environments.append(
                    {
                        "Name": env_name,
                        "Status": env_details.get("Environment", {}).get("Status"),
                        "Arn": env_details.get("Environment", {}).get("Arn"),
                        "CreatedAt": env_details.get("Environment", {}).get("CreatedAt"),
                    }
                )
            except Exception as e:
                logger.error("Error getting details for environment %s: %s", env_name, e)
                environments.append(
                    {
                        "Name": env_name,
                        "Status": "ERROR",
                        "Error": str(e),
                    }
                )

        return {
            "Environments": environments,
            "NextToken": response.get("NextToken"),
        }

    except (ClientError, BotoCoreError) as e:
        logger.error("Error listing environments: %s", e)
        return {"error": str(e)}

awslabs/mwaa_mcp_server/server.py:27-40 (registration)

The registration of the 'list_environments' tool in the FastMCP server, which wraps the `MWAATools.list_environments` method.

@mcp.tool(name="list_environments")
async def list_environments(
    max_results: Optional[int] = None,
) -> Dict[str, Any]:
    """List all MWAA environments in the current AWS account and region.

    Args:
        max_results: Maximum number of environments to return (1-25)

    Returns:
        Dictionary containing list of environment names and metadata
    """
    max_results_int = int(max_results) if max_results is not None else None
    return await tools.list_environments(max_results_int)

MWAA MCP Server