This is page 1 of 2. Use http://codebase.md/alexei-led/aws-mcp-server?format=md&lines=true&page={x} to view the full context.

# Directory Structure

```
├── .dockerignore
├── .github
│   └── workflows
│       ├── ci.yml
│       └── release.yml
├── .gitignore
├── CLAUDE.md
├── codecov.yml
├── deploy
│   └── docker
│       ├── docker-compose.yml
│       └── Dockerfile
├── docs
│   └── VERSION.md
├── LICENSE
├── Makefile
├── media
│   └── demo.mp4
├── pyproject.toml
├── README.md
├── security_config_example.yaml
├── smithery.yaml
├── spec.md
├── src
│   └── aws_mcp_server
│       ├── __init__.py
│       ├── __main__.py
│       ├── cli_executor.py
│       ├── config.py
│       ├── prompts.py
│       ├── resources.py
│       ├── security.py
│       ├── server.py
│       └── tools.py
├── tests
│   ├── __init__.py
│   ├── conftest.py
│   ├── integration
│   │   ├── __init__.py
│   │   ├── test_aws_live.py
│   │   ├── test_security_integration.py
│   │   └── test_server_integration.py
│   ├── test_aws_integration.py
│   ├── test_aws_setup.py
│   ├── test_bucket_creation.py
│   ├── test_run_integration.py
│   └── unit
│       ├── __init__.py
│       ├── test_cli_executor.py
│       ├── test_init.py
│       ├── test_main.py
│       ├── test_prompts.py
│       ├── test_resources.py
│       ├── test_security.py
│       ├── test_server.py
│       └── test_tools.py
└── uv.lock
```

# Files

--------------------------------------------------------------------------------
/.dockerignore:
--------------------------------------------------------------------------------

```
 1 | # Version Control
 2 | .git/
 3 | .github/
 4 | .gitignore
 5 | .gitattributes
 6 | 
 7 | # Docker
 8 | .dockerignore
 9 | deploy/
10 | docker-compose*.yml
11 | Dockerfile*
12 | 
13 | # Documentation
14 | docs/
15 | 
16 | # Markdown files except README.md
17 | *.md
18 | !README.md
19 | 
20 | # Python
21 | __pycache__/
22 | *.py[cod]
23 | *.$py.class
24 | *.so
25 | .Python
26 | *.egg-info/
27 | *.egg
28 | .installed.cfg
29 | build/
30 | develop-eggs/
31 | dist/
32 | downloads/
33 | eggs/
34 | .eggs/
35 | lib/
36 | lib64/
37 | parts/
38 | sdist/
39 | var/
40 | wheels/
41 | 
42 | # Virtual Environments
43 | .env
44 | .venv/
45 | env/
46 | ENV/
47 | venv/
48 | 
49 | # Testing and Coverage
50 | .coverage
51 | .pytest_cache/
52 | .tox/
53 | .nox/
54 | htmlcov/
55 | tests/
56 | 
57 | # Development and IDE
58 | .idea/
59 | .vscode/
60 | .ruff_cache/
61 | .mypy_cache/
62 | .aider*
63 | *.swp
64 | *.swo
65 | 
66 | # OS Generated
67 | .DS_Store
68 | Thumbs.db
69 | 
70 | # Logs
71 | logs/
72 | *.log
```

--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------

```
 1 | # Python
 2 | __pycache__/
 3 | *.py[cod]
 4 | *$py.class
 5 | *.so
 6 | .Python
 7 | build/
 8 | develop-eggs/
 9 | dist/
10 | downloads/
11 | eggs/
12 | .eggs/
13 | lib/
14 | lib64/
15 | parts/
16 | sdist/
17 | var/
18 | wheels/
19 | share/python-wheels/
20 | *.egg-info/
21 | .installed.cfg
22 | *.egg
23 | MANIFEST
24 | 
25 | # Testing and Coverage
26 | .coverage
27 | .coverage.*
28 | .pytest_cache/
29 | .tox/
30 | .nox/
31 | htmlcov/
32 | .hypothesis/
33 | coverage.xml
34 | *.cover
35 | nosetests.xml
36 | 
37 | # Virtual Environments
38 | .env
39 | .venv/
40 | env/
41 | venv/
42 | ENV/
43 | env.bak/
44 | venv.bak/
45 | 
46 | # Development and IDE
47 | .idea/
48 | .vscode/
49 | .ruff_cache/
50 | .mypy_cache/
51 | .dmypy.json
52 | dmypy.json
53 | .pytype/
54 | .spyderproject
55 | .spyproject
56 | .ropeproject
57 | .aider*
58 | *.swp
59 | *.swo
60 | *~
61 | .*.sw[op]
62 | 
63 | # Jupyter
64 | .ipynb_checkpoints
65 | 
66 | # Logs
67 | logs/
68 | *.log
69 | pip-log.txt
70 | pip-delete-this-directory.txt
71 | 
72 | # OS Generated
73 | .DS_Store
74 | Thumbs.db
75 | Icon?
76 | ehthumbs.db
77 | Desktop.ini
78 | 
79 | # Secrets and Credentials
80 | *.pem
81 | *.key
82 | secrets/
83 | config.local.yaml
84 | credentials.json
85 | aws_credentials
86 | 
87 | # Local Development
88 | .direnv/
89 | .envrc
90 | *.local.yml
91 | *.local.yaml
92 | local_settings.py
93 | 
94 | # Distribution
95 | *.tar.gz
96 | *.tgz
97 | *.zip
98 | *.gz
```

--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------

```markdown
  1 | # AWS Model Context Protocol (MCP) Server
  2 | 
  3 | [![CI](https://github.com/alexei-led/aws-mcp-server/actions/workflows/ci.yml/badge.svg)](https://github.com/alexei-led/aws-mcp-server/actions/workflows/ci.yml)
  4 | [![Code Coverage](https://codecov.io/gh/alexei-led/aws-mcp-server/branch/main/graph/badge.svg?token=K8vdP3zyuy)](https://codecov.io/gh/alexei-led/aws-mcp-server)
  5 | [![Linter: Ruff](https://img.shields.io/badge/Linter-Ruff-brightgreen?style=flat-square)](https://github.com/alexei-led/aws-mcp-server)
  6 | [![Image Tags](https://ghcr-badge.egpl.dev/alexei-led/aws-mcp-server/tags?color=%2344cc11&ignore=latest&n=4&label=image+tags&trim=)](https://github.com/alexei-led/aws-mcp-server/pkgs/container/aws-mcp-server/versions)
  7 | [![Image Size](https://ghcr-badge.egpl.dev/alexei-led/aws-mcp-server/size?color=%2344cc11&tag=latest&label=image+size&trim=)](https://github.com/alexei-led/aws-mcp-server/pkgs/container/aws-mcp-server)
  8 | 
  9 | A lightweight service that enables AI assistants to execute AWS CLI commands through the Model Context Protocol (MCP).
 10 | 
 11 | ## Overview
 12 | 
 13 | The AWS MCP Server provides a bridge between MCP-aware AI assistants (like Claude Desktop, Cursor, Windsurf) and the AWS CLI. It enables these assistants to:
 14 | 
 15 | 1. **Retrieve AWS CLI documentation** (`aws_cli_help`) - Get detailed help on AWS services and commands
 16 | 2. **Execute AWS CLI commands** (`aws_cli_pipeline`) - Run commands with Unix pipes and receive formatted results optimized for AI consumption
 17 | 
 18 | ```mermaid
 19 | flowchart LR
 20 |     AI[AI Assistant] <-->|MCP Protocol| Server[AWS MCP Server]
 21 |     Server <-->|Subprocess| AWS[AWS CLI]
 22 |     AWS <-->|API| Cloud[AWS Cloud]
 23 | ```
 24 | 
 25 | ## Demo
 26 | 
 27 | [Demo](https://private-user-images.githubusercontent.com/1898375/424996801-b51ddc8e-5df5-40c4-8509-84c1a7800d62.mp4?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NDI0NzY5OTUsIm5iZiI6MTc0MjQ3NjY5NSwicGF0aCI6Ii8xODk4Mzc1LzQyNDk5NjgwMS1iNTFkZGM4ZS01ZGY1LTQwYzQtODUwOS04NGMxYTc4MDBkNjIubXA0P1gtQW16LUFsZ29yaXRobT1BV1M0LUhNQUMtU0hBMjU2JlgtQW16LUNyZWRlbnRpYWw9QUtJQVZDT0RZTFNBNTNQUUs0WkElMkYyMDI1MDMyMCUyRnVzLWVhc3QtMSUyRnMzJTJGYXdzNF9yZXF1ZXN0JlgtQW16LURhdGU9MjAyNTAzMjBUMTMxODE1WiZYLUFtei1FeHBpcmVzPTMwMCZYLUFtei1TaWduYXR1cmU9NjgwNTM4MDVjN2U4YjQzN2Y2N2Y5MGVkMThiZTgxYWEyNzBhZTlhMTRjZDY3ZDJmMzJkNmViM2U4M2U4MTEzNSZYLUFtei1TaWduZWRIZWFkZXJzPWhvc3QifQ.tIb7uSkDpSaspIluzCliHS8ATmlzkvEnF3CiClD-UGQ)
 28 | 
 29 | The video demonstrates using Claude Desktop with AWS MCP Server to create a new AWS EC2 instance with AWS SSM agent installed.
 30 | 
 31 | ## Features
 32 | 
 33 | - **Command Documentation** - Detailed help information for AWS CLI commands
 34 | - **Command Execution** - Execute AWS CLI commands and return human-readable results
 35 | - **Unix Pipe Support** - Filter and transform AWS CLI output using standard Unix pipes and utilities
 36 | - **AWS Resources Context** - Access to AWS profiles, regions, account information, and environment details via MCP Resources
 37 | - **Prompt Templates** - Pre-defined prompt templates for common AWS tasks following best practices
 38 | - **Docker Integration** - Simple deployment through containerization with multi-architecture support (AMD64/x86_64 and ARM64)
 39 | - **AWS Authentication** - Leverages existing AWS credentials on the host machine
 40 | 
 41 | ## Requirements
 42 | 
 43 | - Docker (default) or Python 3.13+ (and AWS CLI installed locally)
 44 | - AWS credentials configured
 45 | 
 46 | ## Getting Started
 47 | 
 48 | **Note:** For security and reliability, running the server inside a Docker container is the **strongly recommended** method. Please review the [Security Considerations](#security-considerations) section for important considerations.
 49 | 
 50 | ### Run Server Option 1: Using Docker (Recommended)
 51 | 
 52 | ```bash
 53 | # Clone repository
 54 | git clone https://github.com/alexei-led/aws-mcp-server.git
 55 | cd aws-mcp-server
 56 | 
 57 | # Build and run Docker container
 58 | docker compose -f deploy/docker/docker-compose.yml up -d
 59 | ```
 60 | 
 61 | The Docker image supports both AMD64/x86_64 (Intel/AMD) and ARM64 (Apple Silicon M1-M4, AWS Graviton) architectures.
 62 | 
 63 | > **Note**: The official image from GitHub Packages is multi-architecture and will automatically use the appropriate version for your system.
 64 | >
 65 | > ```bash
 66 | > # Use the latest stable version
 67 | > docker pull ghcr.io/alexei-led/aws-mcp-server:latest
 68 | > 
 69 | > # Or pin to a specific version (recommended for production)
 70 | > docker pull ghcr.io/alexei-led/aws-mcp-server:1.0.0
 71 | > ```
 72 | >
 73 | > **Docker Image Tags**:
 74 | >
 75 | > - `latest`: Latest stable release
 76 | > - `x.y.z` (e.g., `1.0.0`): Specific version
 77 | > - `sha-<commit-sha>`: Development builds, tagged with Git commit SHA (e.g., `sha-gb697684`)
 78 | 
 79 | ### Run Server Option 2: Using Python
 80 | 
 81 | **Use with Caution:** Running natively requires careful environment setup and carries higher security risks compared to the recommended Docker deployment. Ensure you understand the implications outlined in the [Security Considerations](#security-considerations) section.
 82 | 
 83 | ```bash
 84 | # Clone repository
 85 | git clone https://github.com/alexei-led/aws-mcp-server.git
 86 | cd aws-mcp-server
 87 | 
 88 | # Set up virtual environment
 89 | python -m venv venv
 90 | source venv/bin/activate  # On Windows: venv\Scripts\activate
 91 | 
 92 | # Install in development mode
 93 | pip install -e .
 94 | 
 95 | # Run the server
 96 | python -m aws_mcp_server
 97 | ```
 98 | 
 99 | ## Configuration
100 | 
101 | The AWS MCP Server can be configured using environment variables:
102 | 
103 | | Environment Variable      | Description                                  | Default   |
104 | |--------------------------|----------------------------------------------|-----------|
105 | | `AWS_MCP_TIMEOUT`        | Command execution timeout in seconds         | 300       |
106 | | `AWS_MCP_MAX_OUTPUT`     | Maximum output size in characters            | 100000    |
107 | | `AWS_MCP_TRANSPORT`      | Transport protocol to use ("stdio" or "sse") | stdio     |
108 | | `AWS_PROFILE`            | AWS profile to use                           | default   |
109 | | `AWS_REGION`             | AWS region to use                            | us-east-1 |
110 | | `AWS_MCP_SECURITY_MODE`  | Security mode ("strict" or "permissive")     | strict    |
111 | | `AWS_MCP_SECURITY_CONFIG`| Path to custom security configuration file   | ""        |
112 | 
113 | **Important:** Securely manage the AWS credentials provided to the server, whether via mounted `~/.aws` files or environment variables. Ensure the credentials follow the principle of least privilege as detailed in the [Security Considerations](#security-considerations) section. When running via Docker, ensure these variables are passed correctly to the container environment (e.g., using `docker run -e VAR=value ...`).
114 | 
115 | ## Security Considerations
116 | 
117 | Security is paramount when executing commands against your AWS environment. While AWS MCP Server provides functionality, **you are responsible** for configuring and running it securely. Please adhere strictly to the following:
118 | 
119 | **1. Recommended Deployment: Docker Container**
120 | 
121 | *   **Isolation:** Running the server inside a Docker container is the **strongly recommended and default** deployment method. Containerization provides crucial filesystem and process isolation. Potentially destructive Unix commands (like `rm`, `mv`) executed via pipes, even if misused, will be contained within the ephemeral Docker environment and will **not** affect your host machine's filesystem. The container can be easily stopped and recreated.
122 | *   **Controlled Environment:** Docker ensures a consistent environment with necessary dependencies, reducing unexpected behavior.
123 | 
124 | **2. AWS Credentials and IAM Least Privilege (Critical)**
125 | 
126 | *   **User Responsibility:** You provide the AWS credentials to the server (via mounted `~/.aws` or environment variables).
127 | *   **Least Privilege is Essential:** The server executes AWS CLI commands *using the credentials you provide*. It is **absolutely critical** that these credentials belong to an IAM principal (User or Role) configured with the **minimum necessary permissions** (least privilege) for *only* the AWS actions you intend to perform through this tool.
128 |     *   **Do Not Use Root Credentials:** Never use AWS account root user credentials.
129 |     *   **Regularly Review Permissions:** Periodically audit the IAM permissions associated with the credentials.
130 | *   **Impact Limitation:** Properly configured IAM permissions are the **primary mechanism** for limiting the potential impact of *any* command executed via the server, whether intended or unintended. Even if a command were manipulated, it could only perform actions allowed by the specific IAM policy.
131 | 
132 | **3. Trusted User Model**
133 | 
134 | *   The server assumes the end-user interacting with the MCP client (e.g., Claude Desktop, Cursor) is the **same trusted individual** who configured the server and provided the least-privilege AWS credentials. Do not expose the server or connected client to untrusted users.
135 | 
136 | **4. Understanding Execution Risks (Current Implementation)**
137 | 
138 | *   **Command Execution:** The current implementation uses shell features (`shell=True` in subprocess calls) to execute AWS commands and handle Unix pipes. While convenient, this approach carries inherent risks if the input command string were manipulated (command injection).
139 | *   **Mitigation via Operational Controls:** In the context of the **trusted user model** and **Docker deployment**, these risks are mitigated operationally:
140 |     *   The trusted user is assumed not to provide intentionally malicious commands against their own environment.
141 |     *   Docker contains filesystem side-effects.
142 |     *   **Crucially, IAM least privilege limits the scope of *any* AWS action that could be executed.**
143 | *   **Credential Exfiltration Risk:** Despite containerization and IAM, a sophisticated command injection could potentially attempt to read the mounted credentials (`~/.aws`) or environment variables within the container and exfiltrate them (e.g., via `curl`). **Strict IAM policies remain the most vital defense** to limit the value of potentially exfiltrated credentials.
144 | 
145 | **5. Network Exposure (SSE Transport)**
146 | 
147 | *   If using the `sse` transport (which implies a network listener), ensure you bind the server only to trusted network interfaces (e.g., `localhost`) or implement appropriate network security controls (firewalls, authentication proxies) if exposing it more broadly. The default `stdio` transport does not open network ports.
148 | 
149 | **6. Shared Responsibility Summary**
150 | 
151 | *   **AWS MCP Server provides the tool.**
152 | *   **You, the user, are responsible for:**
153 |     *   Running it within the recommended secure Docker environment.
154 |     *   Providing and securely managing **least-privilege** AWS credentials.
155 |     *   Ensuring only trusted users interact with the server/client.
156 |     *   Securing the network environment if applicable.
157 | 
158 | By strictly adhering to Docker deployment and meticulous IAM least-privilege configuration, you establish the necessary operational controls for using the AWS MCP Server securely with its current implementation.
159 | 
160 | ## Integrating with Claude Desktop
161 | 
162 | ### Configuration
163 | 
164 | To manually integrate AWS MCP Server with Claude Desktop:
165 | 
166 | 1. **Locate the Claude Desktop configuration file**:
167 |    - macOS: `~/Library/Application Support/Claude/claude_desktop_config.json`
168 |    - Windows: `%APPDATA%\Claude\claude_desktop_config.json`
169 | 
170 | 2. **Edit the configuration file** to include the AWS MCP Server:
171 |    ```json
172 |    {
173 |      "mcpServers": {
174 |        "aws-mcp-server": {
175 |          "command": "docker",
176 |          "args": [
177 |            "run",
178 |            "-i",
179 |            "--rm",
180 |            "-v",
181 |            "/Users/YOUR_USER_NAME/.aws:/home/appuser/.aws:ro",
182 |            "ghcr.io/alexei-led/aws-mcp-server:latest"
183 |          ]
184 |        }
185 |      }
186 |    }
187 |    ```
188 |    
189 | 3. **Restart Claude Desktop** to apply the changes
190 |    - After restarting, you should see a hammer 🔨 icon in the bottom right corner of the input box
191 |    - This indicates that the AWS MCP Server is available for use
192 | 
193 | ```mermaid
194 | flowchart TD
195 |     subgraph "User Device"
196 |         config[Edit claude_desktop_config.json]
197 |         claude[Claude Desktop]
198 |         docker[Docker Container]
199 |         aws_creds[AWS Credentials]
200 |     end
201 |     
202 |     subgraph "AWS Cloud"
203 |         services[AWS Services]
204 |     end
205 |     
206 |     subgraph "MCP Server Components"
207 |         tools[AWS CLI Tools]
208 |         resources[AWS Resources]
209 |         templates[Prompt Templates]
210 |     end
211 |     
212 |     config -->|Add MCP Server Config| claude
213 |     claude -->|Docker Run Command| docker
214 |     docker --- tools
215 |     docker --- resources
216 |     docker --- templates
217 |     aws_creds -->|Mount Read-only| docker
218 |     resources -.->|Read| aws_creds
219 |     tools -->|API Calls| services
220 | ```
221 | 
222 | ### Example Interactions
223 | 
224 | **Getting AWS CLI Documentation**:
225 | ```
226 | User: What are the available options for the AWS S3 cp command?
227 | 
228 | Claude: Let me get that information for you.
229 | 
230 | The AWS S3 cp command has the following options:
231 | 
232 | NAME
233 |     cp - Copies a local file or S3 object to another location locally or in S3.
234 | 
235 | SYNOPSIS
236 |     cp <LocalPath> <S3Uri> or <S3Uri> <LocalPath> or <S3Uri> <S3Uri> [--acl <value>] [--cache-control <value>] ...
237 | ```
238 | 
239 | **Executing AWS CLI Commands**:
240 | ```
241 | User: List my S3 buckets
242 | 
243 | Claude: I'll list your S3 buckets for you.
244 | 
245 | 2023-10-15 14:30:45 my-bucket-1
246 | 2023-11-20 09:15:32 my-bucket-2
247 | 2024-01-05 11:22:18 my-backup-bucket
248 | ```
249 | 
250 | **Using Command Pipes**:
251 | ```
252 | User: List my S3 buckets that have "backup" in their name and show them in alphabetical order
253 | 
254 | Claude: Let me find that for you.
255 | 
256 | 2024-01-05 11:22:18 my-backup-bucket
257 | 2024-03-01 09:44:12 weekly-backup-bucket
258 | 2024-03-15 13:10:57 database-backup-bucket
259 | ```
260 | 
261 | **Accessing AWS Resources**:
262 | ```
263 | User: What AWS regions are available for me to use?
264 | 
265 | Claude: Let me check what AWS regions are available for you.
266 | 
267 | Available regions:
268 | - us-east-1 (US East, N. Virginia) - Currently selected
269 | - us-east-2 (US East, Ohio)
270 | - us-west-1 (US West, N. California)
271 | - us-west-2 (US West, Oregon)
272 | - eu-west-1 (EU West, Ireland)
273 | - eu-central-1 (EU Central, Frankfurt)
274 | - ap-northeast-1 (Asia Pacific, Tokyo)
275 | - ap-southeast-1 (Asia Pacific, Singapore)
276 | - ap-southeast-2 (Asia Pacific, Sydney)
277 | 
278 | You're currently using the us-east-1 region.
279 | ```
280 | 
281 | **Using Prompt Templates**:
282 | ```
283 | User: I need to perform a security audit of my S3 buckets
284 | 
285 | Claude: I'll help you audit the security of your S3 buckets.
286 | 
287 | Here are AWS CLI commands to perform a comprehensive security audit:
288 | 
289 | # List all buckets with public access settings
290 | aws s3api list-buckets --query 'Buckets[*].Name' --output text | xargs -I {} aws s3api get-public-access-block --bucket {}
291 | 
292 | # Check bucket policies for public access
293 | aws s3api list-buckets --query 'Buckets[*].Name' --output text | xargs -I {} aws s3api get-bucket-policy --bucket {} 2>/dev/null
294 | 
295 | # Identify buckets without encryption
296 | aws s3api list-buckets --query 'Buckets[*].Name' --output text | xargs -I {} aws s3api get-bucket-encryption --bucket {} 2>/dev/null
297 | 
298 | # Check for buckets without logging enabled
299 | aws s3api list-buckets --query 'Buckets[*].Name' --output text | xargs -I {} aws s3api get-bucket-logging --bucket {} 2>/dev/null
300 | 
301 | # Verify bucket versioning status
302 | aws s3api list-buckets --query 'Buckets[*].Name' --output text | xargs -I {} aws s3api get-bucket-versioning --bucket {}
303 | ```
304 | 
305 | ## Available Prompt Templates
306 | 
307 | The AWS MCP Server includes the following pre-defined prompt templates:
308 | 
309 | ### Core Operations
310 | 
311 | | Prompt                 | Description                                                   | Parameters                                          |
312 | |------------------------|---------------------------------------------------------------|-----------------------------------------------------|
313 | | `create_resource`      | Generate commands to create AWS resources with best practices | `resource_type`, `resource_name`                    |
314 | | `resource_inventory`   | Create comprehensive inventory of resources                   | `service`, `region` (optional)                      |
315 | | `troubleshoot_service` | Generate commands to troubleshoot service issues              | `service`, `resource_id`                            |
316 | | `resource_cleanup`     | Identify and safely clean up resources                        | `service`, `criteria` (optional)                    |
317 | 
318 | ### Security & Compliance
319 | 
320 | | Prompt                     | Description                                                | Parameters                                          |
321 | |----------------------------|------------------------------------------------------------|-----------------------------------------------------|
322 | | `security_audit`           | Audit security settings for a specific AWS service         | `service`                                           |
323 | | `security_posture_assessment` | Comprehensive security assessment across your AWS environment | None                                          |
324 | | `iam_policy_generator`     | Create least-privilege IAM policies                        | `service`, `actions`, `resource_pattern` (optional) |
325 | | `compliance_check`         | Check compliance with standards                            | `compliance_standard`, `service` (optional)         |
326 | 
327 | ### Cost & Performance
328 | 
329 | | Prompt               | Description                                             | Parameters                                         |
330 | |----------------------|---------------------------------------------------------|----------------------------------------------------|
331 | | `cost_optimization`  | Find cost optimization opportunities for a service      | `service`                                          |
332 | | `performance_tuning` | Optimize and tune performance of AWS resources          | `service`, `resource_id`                           |
333 | 
334 | ### Infrastructure & Architecture
335 | 
336 | | Prompt                      | Description                                              | Parameters                                           |
337 | |-----------------------------|----------------------------------------------------------|------------------------------------------------------|
338 | | `serverless_deployment`     | Deploy serverless applications with best practices       | `application_name`, `runtime` (optional)             |
339 | | `container_orchestration`   | Set up container environments (ECS/EKS)                  | `cluster_name`, `service_type` (optional)            |
340 | | `vpc_network_design`        | Design and implement secure VPC networking               | `vpc_name`, `cidr_block` (optional)                  |
341 | | `infrastructure_automation` | Automate infrastructure management                       | `resource_type`, `automation_scope` (optional)       |
342 | | `multi_account_governance`  | Implement secure multi-account strategies                | `account_type` (optional)                            |
343 | 
344 | ### Reliability & Monitoring
345 | 
346 | | Prompt               | Description                                           | Parameters                                          |
347 | |----------------------|-------------------------------------------------------|-----------------------------------------------------|
348 | | `service_monitoring` | Set up comprehensive monitoring                       | `service`, `metric_type` (optional)                 |
349 | | `disaster_recovery`  | Implement enterprise-grade DR solutions               | `service`, `recovery_point_objective` (optional)    |
350 | 
351 | ## Security
352 | 
353 | The AWS MCP Server implements a comprehensive multi-layered approach to command validation and security:
354 | 
355 | ### Command Validation System
356 | 
357 | The server validates all AWS CLI commands through a three-layer system:
358 | 
359 | 1. **Basic Command Structure**: 
360 |    - Verifies commands start with 'aws' prefix and contain a valid service
361 |    - Ensures proper command syntax
362 | 
363 | 2. **Security-Focused Command Filtering**:
364 |    - **Dangerous Commands**: Blocks commands that could compromise security
365 |    - **Safe Patterns**: Explicitly allows read-only operations needed for normal use
366 |    - **Regex Pattern Matching**: Prevents complex security risks with pattern matching
367 | 
368 | 3. **Pipe Command Security**:
369 |    - Validates Unix commands used in pipes
370 |    - Restricts commands to a safe allowlist
371 |    - Prevents filesystem manipulation and arbitrary command execution
372 | 
373 | ### Default Security Configuration
374 | 
375 | The default security configuration focuses on preventing the following attack vectors:
376 | 
377 | #### 1. Identity and Access Management (IAM) Risks
378 | 
379 | | Blocked Command | Security Risk |
380 | |-----------------|---------------|
381 | | `aws iam create-user` | Creates potential backdoor accounts with persistent access |
382 | | `aws iam create-access-key` | Creates long-term credentials that can be stolen or misused |
383 | | `aws iam attach-*-policy` | Potential privilege escalation via policy attachments |
384 | | `aws iam put-user-policy` | Inline policies can grant excessive permissions |
385 | | `aws iam create-policy` | Creating new policies with potentially dangerous permissions |
386 | | `aws iam create-login-profile` | Creates console passwords for existing users |
387 | | `aws iam deactivate-mfa-device` | Disables multi-factor authentication, weakening security |
388 | | `aws iam update-assume-role-policy` | Modifies trust relationships, enabling privilege escalation |
389 | 
390 | #### 2. Audit and Logging Tampering
391 | 
392 | | Blocked Command | Security Risk |
393 | |-----------------|---------------|
394 | | `aws cloudtrail delete-trail` | Removes audit trail of AWS activity |
395 | | `aws cloudtrail stop-logging` | Stops collecting activity logs, creating blind spots |
396 | | `aws cloudtrail update-trail` | Can redirect or modify logging configuration |
397 | | `aws config delete-configuration-recorder` | Disables AWS Config recording of resource changes |
398 | | `aws guardduty delete-detector` | Disables threat detection capabilities |
399 | 
400 | #### 3. Sensitive Data Access and Protection
401 | 
402 | | Blocked Command | Security Risk |
403 | |-----------------|---------------|
404 | | `aws secretsmanager put-secret-value` | Modifies sensitive credentials |
405 | | `aws secretsmanager delete-secret` | Removes sensitive credentials |
406 | | `aws kms schedule-key-deletion` | Schedules deletion of encryption keys, risking data loss |
407 | | `aws kms disable-key` | Disables encryption keys, potentially exposing data |
408 | | `aws s3api put-bucket-policy` | Can create public S3 buckets, exposing data |
409 | | `aws s3api delete-bucket-policy` | Removes protective policies from buckets |
410 | 
411 | #### 4. Network Security Risks
412 | 
413 | | Blocked Command | Security Risk |
414 | |-----------------|---------------|
415 | | `aws ec2 authorize-security-group-ingress` | Opens inbound network access, potential exposure |
416 | | `aws ec2 authorize-security-group-egress` | Opens outbound network access, potential data exfiltration |
417 | | `aws ec2 modify-instance-attribute` | Can alter security properties of instances |
418 | 
419 | Many read-only operations that match these patterns are explicitly allowed via safe patterns:
420 | 
421 | - All `get-`, `list-`, and `describe-` commands
422 | - All help commands (`--help`, `help`)
423 | - Simulation and testing commands (e.g., `aws iam simulate-custom-policy`)
424 | 
425 | ### Configuration Options
426 | 
427 | - **Security Modes**:
428 |   - `strict` (default): Enforces all security validations
429 |   - `permissive`: Logs warnings but allows execution (use with caution)
430 | 
431 | - **Custom Configuration**:
432 |   - Override default security rules via YAML configuration file
433 |   - Configure service-specific dangerous commands
434 |   - Define custom safe patterns and regex rules
435 |   - Environment variable: `AWS_MCP_SECURITY_CONFIG`
436 | 
437 | - **Execution Controls**:
438 |   - Timeouts prevent long-running commands (default: 300 seconds)
439 |   - Output size limits prevent memory issues
440 |   - Environment variables: `AWS_MCP_TIMEOUT`, `AWS_MCP_MAX_OUTPUT`
441 | 
442 | ### Custom Security Rules Example
443 | 
444 | You can create custom security rules by defining a YAML configuration file:
445 | 
446 | ```yaml
447 | # Example custom security configuration
448 | # Save to a file and set AWS_MCP_SECURITY_CONFIG environment variable
449 | 
450 | # Dangerous commands to block
451 | dangerous_commands:
452 |   iam:
453 |     # Only block specific IAM operations for your environment
454 |     - "aws iam create-user"
455 |     - "aws iam attach-user-policy"
456 |   
457 |   # Custom service restrictions for your organization
458 |   lambda:
459 |     - "aws lambda delete-function"
460 |     - "aws lambda remove-permission"
461 |   
462 |   # Prevent accidental DynamoDB table deletion
463 |   dynamodb:
464 |     - "aws dynamodb delete-table"
465 | 
466 | # Safe patterns to explicitly allow
467 | safe_patterns:
468 |   # Global safe patterns
469 |   general:
470 |     - "--help"
471 |     - "--dry-run"
472 |   
473 |   # Allow read operations on IAM
474 |   iam:
475 |     - "aws iam get-"
476 |     - "aws iam list-"
477 |   
478 |   # Allow specific Lambda operations
479 |   lambda:
480 |     - "aws lambda list-functions"
481 |     - "aws lambda get-function"
482 | 
483 | # Complex regex rules for security validation
484 | regex_rules:
485 |   general:
486 |     # Prevent use of root credentials
487 |     - pattern: "aws .* --profile\\s+root"
488 |       description: "Prevent use of root profile"
489 |       error_message: "Using the root profile is not allowed for security reasons"
490 |   
491 |   iam:
492 |     # Block creation of admin users
493 |     - pattern: "aws iam create-user.*--user-name\\s+.*admin.*"
494 |       description: "Prevent creation of admin users"
495 |       error_message: "Creating users with 'admin' in the name is restricted"
496 |     
497 |     # Prevent wildcards in IAM policies
498 |     - pattern: "aws iam create-policy.*\"Effect\":\\s*\"Allow\".*\"Action\":\\s*\"\\*\".*\"Resource\":\\s*\"\\*\""
499 |       description: "Prevent wildcards in policies"
500 |       error_message: "Creating policies with '*' wildcards for both Action and Resource is not allowed"
501 |   
502 |   s3:
503 |     # Prevent public bucket policies
504 |     - pattern: "aws s3api put-bucket-policy.*\"Effect\":\\s*\"Allow\".*\"Principal\":\\s*\"\\*\""
505 |       description: "Prevent public bucket policies"
506 |       error_message: "Creating bucket policies with public access is restricted"
507 | ```
508 | 
509 | ### Security Examples
510 | 
511 | The system follows IAM best practices, focusing on preventing escalation of privilege:
512 | 
513 | ```bash
514 | # This command would be blocked (creates user)
515 | aws iam create-user --user-name new-user
516 | > Error: This command (aws iam create-user) is restricted for security reasons.
517 | 
518 | # This command would be blocked (attaches admin policy)
519 | aws iam attach-user-policy --user-name any-user --policy-arn arn:aws:iam::aws:policy/AdministratorAccess
520 | > Error: Attaching Administrator policies is restricted for security reasons.
521 | 
522 | # This command would be blocked (opens SSH port globally)
523 | aws ec2 authorize-security-group-ingress --group-id sg-12345 --protocol tcp --port 22 --cidr 0.0.0.0/0
524 | > Error: Opening non-web ports to the entire internet (0.0.0.0/0) is restricted.
525 | 
526 | # These commands are allowed (read-only operations)
527 | aws iam list-users
528 | aws s3 ls
529 | aws ec2 describe-instances
530 | ```
531 | 
532 | ### Security Best Practices
533 | 
534 | - Always use the default `strict` security mode in production
535 | - Follow the deployment recommendations in [Security Considerations](#security-considerations)
536 | - Run with least-privilege AWS credentials
537 | - For custom configurations, focus on your security requirements
538 | 
539 | ## Development
540 | 
541 | ### Setting Up the Development Environment
542 | 
543 | ```bash
544 | # Install only runtime dependencies using pip
545 | pip install -e .
546 | 
547 | # Install all development dependencies using pip
548 | pip install -e ".[dev]"
549 | 
550 | # Or use uv for faster dependency management
551 | make uv-install       # Install runtime dependencies
552 | make uv-dev-install   # Install development dependencies
553 | ```
554 | 
555 | ### Makefile Commands
556 | 
557 | The project includes a Makefile with various targets for common tasks:
558 | 
559 | ```bash
560 | # Test commands
561 | make test             # Run tests excluding integration tests
562 | make test-unit        # Run unit tests only (all tests except integration tests)
563 | make test-integration # Run integration tests only (requires AWS credentials)
564 | make test-all         # Run all tests including integration tests
565 | 
566 | # Test coverage commands
567 | make test-coverage    # Run tests with coverage report (excluding integration tests)
568 | make test-coverage-all # Run all tests with coverage report (including integration tests)
569 | 
570 | # Linting and formatting
571 | make lint             # Run linters (ruff check and format --check)
572 | make lint-fix         # Run linters and auto-fix issues where possible
573 | make format           # Format code with ruff
574 | ```
575 | 
576 | For a complete list of available commands, run `make help`.
577 | 
578 | ### Code Coverage
579 | 
580 | The project includes configuration for [Codecov](https://codecov.io) to track code coverage metrics. The configuration is in the `codecov.yml` file, which:
581 | 
582 | - Sets a target coverage threshold of 80%
583 | - Excludes test files, setup files, and documentation from coverage reports
584 | - Configures PR comments and status checks
585 | 
586 | Coverage reports are automatically generated during CI/CD runs and uploaded to Codecov.
587 | 
588 | ### Integration Testing
589 | 
590 | Integration tests verify AWS MCP Server works correctly with actual AWS resources. To run them:
591 | 
592 | 1. **Set up AWS resources**:
593 |    - Create an S3 bucket for testing
594 |    - Set the environment variable: `export AWS_TEST_BUCKET=your-test-bucket-name`
595 |    - Ensure your AWS credentials are configured
596 | 
597 | 2. **Run integration tests**:
598 |    ```bash
599 |    # Run all tests including integration tests
600 |    make test-all
601 |    
602 |    # Run only integration tests
603 |    make test-integration
604 |    ```
605 | 
606 | Or you can run the pytest commands directly:
607 | ```bash
608 | # Run all tests including integration tests
609 | pytest --run-integration
610 | 
611 | # Run only integration tests
612 | pytest --run-integration -m integration
613 | ```
614 | 
615 | ## Troubleshooting
616 | 
617 | - **Authentication Issues**: Ensure your AWS credentials are properly configured
618 | - **Connection Errors**: Verify the server is running and AI assistant connection settings are correct
619 | - **Permission Errors**: Check that your AWS credentials have the necessary permissions
620 | - **Timeout Errors**: For long-running commands, increase the `AWS_MCP_TIMEOUT` environment variable
621 | 
622 | ## Why Deploy with Docker
623 | 
624 | Deploying AWS MCP Server via Docker is the recommended approach, offering significant security and reliability advantages that form the core of the tool's secure usage pattern:
625 | 
626 | ### Security Benefits
627 | 
628 | - **Isolation (Primary Mitigation):** The Docker container provides essential filesystem and process isolation. AWS CLI commands and piped Unix utilities run in a contained environment. Accidental or misused commands affecting the filesystem are limited to the container, **protecting your host machine**.
629 | - **Controlled Credential Access:** When mounting credentials, using the `:ro` (read-only) flag limits the container's ability to modify your AWS configuration files.
630 | - **No Local Installation:** Avoids installing the AWS CLI and its dependencies directly on your host system.
631 | - **Clean Environment:** Each container run starts with a known, clean state.
632 | 
633 | ### Reliability Advantages
634 | 
635 | - **Consistent Configuration**: All required tools (AWS CLI, SSM plugin, jq) are pre-installed and properly configured
636 | - **Dependency Management**: Avoid version conflicts between tools and dependencies
637 | - **Cross-Platform Consistency**: Works the same way across different operating systems
638 | - **Complete Environment**: Includes all necessary tools for command pipes, filtering, and formatting
639 | 
640 | ### Other Benefits
641 | 
642 | - **Multi-Architecture Support**: Runs on both Intel/AMD (x86_64) and ARM (Apple Silicon, AWS Graviton) processors
643 | - **Simple Updates**: Update to new versions with a single pull command
644 | - **No Python Environment Conflicts**: Avoids potential conflicts with other Python applications on your system
645 | - **Version Pinning**: Easily pin to specific versions for stability in production environments
646 | 
647 | ## Versioning
648 | 
649 | This project uses [setuptools_scm](https://github.com/pypa/setuptools_scm) to automatically determine versions based on Git tags:
650 | 
651 | - **Release versions**: When a Git tag exists (e.g., `1.2.3`), the version will be exactly that tag
652 | - **Development versions**: For commits without tags, a development version is generated in the format: 
653 |   `<last-tag>.post<commits-since-tag>+g<commit-hash>.d<date>` (e.g., `1.2.3.post10+gb697684.d20250406`)
654 | 
655 | The version is automatically included in:
656 | - Package version information
657 | - Docker image labels
658 | - Continuous integration builds
659 | 
660 | ### Creating Releases
661 | 
662 | To create a new release version:
663 | 
664 | ```bash
665 | # Create and push a new tag
666 | git tag -a 1.2.3 -m "Release version 1.2.3"
667 | git push origin 1.2.3
668 | ```
669 | 
670 | The CI/CD pipeline will automatically build and publish Docker images with appropriate version tags.
671 | 
672 | For more detailed information about the version management system, see [VERSION.md](docs/VERSION.md).
673 | 
674 | ## License
675 | 
676 | This project is licensed under the MIT License - see the LICENSE file for details.
677 | 
```

--------------------------------------------------------------------------------
/CLAUDE.md:
--------------------------------------------------------------------------------

```markdown
 1 | # AWS MCP Server Development Guide
 2 | 
 3 | ## Build & Test Commands
 4 | 
 5 | ### Using uv (recommended)
 6 | - Install dependencies: `uv pip install --system -e .`
 7 | - Install dev dependencies: `uv pip install --system -e ".[dev]"`
 8 | - Update lock file: `uv pip compile --system pyproject.toml -o uv.lock`
 9 | - Install from lock file: `uv pip sync --system uv.lock`
10 | 
11 | ### Using pip (alternative)
12 | - Install dependencies: `pip install -e .`
13 | - Install dev dependencies: `pip install -e ".[dev]"`
14 | 
15 | ### Running the server
16 | - Run server: `python -m aws_mcp_server`
17 | - Run server with SSE transport: `AWS_MCP_TRANSPORT=sse python -m aws_mcp_server`
18 | - Run with MCP CLI: `mcp run src/aws_mcp_server/server.py`
19 | 
20 | ### Testing and linting
21 | - Run tests: `pytest`
22 | - Run single test: `pytest tests/path/to/test_file.py::test_function_name -v`
23 | - Run tests with coverage: `python -m pytest --cov=src/aws_mcp_server tests/`
24 | - Run linter: `ruff check src/ tests/`
25 | - Format code: `ruff format src/ tests/`
26 | 
27 | ## Technical Stack
28 | 
29 | - **Python version**: Python 3.13+
30 | - **Project config**: `pyproject.toml` for configuration and dependency management
31 | - **Environment**: Use virtual environment in `.venv` for dependency isolation
32 | - **Package management**: Use `uv` for faster, more reliable dependency management with lock file
33 | - **Dependencies**: Separate production and dev dependencies in `pyproject.toml`
34 | - **Version management**: Use `setuptools_scm` for automatic versioning from Git tags
35 | - **Linting**: `ruff` for style and error checking
36 | - **Type checking**: Use VS Code with Pylance for static type checking
37 | - **Project layout**: Organize code with `src/` layout
38 | 
39 | ## Code Style Guidelines
40 | 
41 | - **Formatting**: Black-compatible formatting via `ruff format`
42 | - **Imports**: Sort imports with `ruff` (stdlib, third-party, local)
43 | - **Type hints**: Use native Python type hints (e.g., `list[str]` not `List[str]`)
44 | - **Documentation**: Google-style docstrings for all modules, classes, functions
45 | - **Naming**: snake_case for variables/functions, PascalCase for classes
46 | - **Function length**: Keep functions short (< 30 lines) and single-purpose
47 | - **PEP 8**: Follow PEP 8 style guide (enforced via `ruff`)
48 | 
49 | ## Python Best Practices
50 | 
51 | - **File handling**: Prefer `pathlib.Path` over `os.path`
52 | - **Debugging**: Use `logging` module instead of `print`
53 | - **Error handling**: Use specific exceptions with context messages and proper logging
54 | - **Data structures**: Use list/dict comprehensions for concise, readable code
55 | - **Function arguments**: Avoid mutable default arguments
56 | - **Data containers**: Leverage `dataclasses` to reduce boilerplate
57 | - **Configuration**: Use environment variables (via `python-dotenv`) for configuration
58 | - **AWS CLI**: Validate all commands before execution (must start with "aws")
59 | - **Security**: Never store/log AWS credentials, set command timeouts
60 | 
61 | ## Development Patterns & Best Practices
62 | 
63 | - **Favor simplicity**: Choose the simplest solution that meets requirements
64 | - **DRY principle**: Avoid code duplication; reuse existing functionality
65 | - **Configuration management**: Use environment variables for different environments
66 | - **Focused changes**: Only implement explicitly requested or fully understood changes
67 | - **Preserve patterns**: Follow existing code patterns when fixing bugs
68 | - **File size**: Keep files under 300 lines; refactor when exceeding this limit
69 | - **Test coverage**: Write comprehensive unit and integration tests with `pytest`; include fixtures
70 | - **Test structure**: Use table-driven tests with parameterization for similar test cases
71 | - **Mocking**: Use unittest.mock for external dependencies; don't test implementation details
72 | - **Modular design**: Create reusable, modular components
73 | - **Logging**: Implement appropriate logging levels (debug, info, error)
74 | - **Error handling**: Implement robust error handling for production reliability
75 | - **Security best practices**: Follow input validation and data protection practices
76 | - **Performance**: Optimize critical code sections when necessary
77 | - **Dependency management**: Add libraries only when essential
78 |   - When adding/updating dependencies, update `pyproject.toml` first
79 |   - Regenerate the lock file with `uv pip compile --system pyproject.toml -o uv.lock`
80 |   - Install the new dependencies with `uv pip sync --system uv.lock`
81 | 
82 | ## Development Workflow
83 | 
84 | - **Version control**: Commit frequently with clear messages
85 | - **Versioning**: Use Git tags for versioning (e.g., `git tag -a 1.2.3 -m "Release 1.2.3"`)
86 |   - For releases, create and push a tag
87 |   - For development, let `setuptools_scm` automatically determine versions
88 | - **Impact assessment**: Evaluate how changes affect other codebase areas
89 | - **Documentation**: Keep documentation up-to-date for complex logic and features
90 | - **Dependencies**: When adding dependencies, always update the `uv.lock` file
91 | - **CI/CD**: All changes should pass CI checks (tests, linting, etc.) before merging
92 | 
```

--------------------------------------------------------------------------------
/tests/unit/__init__.py:
--------------------------------------------------------------------------------

```python
1 | """Unit tests for AWS MCP Server."""
2 | 
```

--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------

```python
1 | """Test package for AWS MCP Server."""
2 | 
```

--------------------------------------------------------------------------------
/tests/integration/__init__.py:
--------------------------------------------------------------------------------

```python
1 | """Integration tests for AWS MCP Server."""
2 | 
```

--------------------------------------------------------------------------------
/tests/test_run_integration.py:
--------------------------------------------------------------------------------

```python
 1 | """Simple test to verify integration test setup."""
 2 | 
 3 | import pytest
 4 | 
 5 | 
 6 | @pytest.mark.integration
 7 | def test_integration_marker_works():
 8 |     """Test that tests with integration marker run."""
 9 |     print("Integration test is running!")
10 |     assert True
11 | 
```

--------------------------------------------------------------------------------
/src/aws_mcp_server/__init__.py:
--------------------------------------------------------------------------------

```python
 1 | """AWS Model Context Protocol (MCP) Server.
 2 | 
 3 | A lightweight service that enables AI assistants to execute AWS CLI commands through the Model Context Protocol (MCP).
 4 | """
 5 | 
 6 | from importlib.metadata import PackageNotFoundError, version
 7 | 
 8 | try:
 9 |     __version__ = version("aws-mcp-server")
10 | except PackageNotFoundError:
11 |     # package is not installed
12 |     pass
13 | 
```

--------------------------------------------------------------------------------
/tests/test_aws_integration.py:
--------------------------------------------------------------------------------

```python
 1 | """Simple test to verify AWS integration setup."""
 2 | 
 3 | import pytest
 4 | 
 5 | 
 6 | @pytest.mark.integration
 7 | def test_aws_credentials(ensure_aws_credentials):
 8 |     """Test that AWS credentials fixture works."""
 9 |     print("AWS credentials test is running!")
10 |     assert True
11 | 
12 | 
13 | @pytest.mark.integration
14 | @pytest.mark.asyncio
15 | async def test_aws_bucket(aws_s3_bucket):
16 |     """Test that AWS bucket fixture works."""
17 |     # We need to manually extract the bucket name from the async generator
18 |     bucket_name = None
19 |     async for name in aws_s3_bucket:
20 |         bucket_name = name
21 |         break
22 | 
23 |     print(f"AWS bucket fixture returned: {bucket_name}")
24 |     assert bucket_name is not None
25 |     assert isinstance(bucket_name, str)
26 |     assert len(bucket_name) > 0
27 | 
```

--------------------------------------------------------------------------------
/codecov.yml:
--------------------------------------------------------------------------------

```yaml
 1 | codecov:
 2 |   require_ci_to_pass: yes
 3 |   notify:
 4 |     wait_for_ci: yes
 5 | 
 6 | coverage:
 7 |   precision: 2
 8 |   round: down
 9 |   range: "70...90"
10 |   status:
11 |     project:
12 |       default:
13 |         # Target minimum coverage percentage
14 |         target: 80%
15 |         # Allow a small decrease in coverage without failing
16 |         threshold: 5%
17 |         if_ci_failed: error
18 |     patch:
19 |       default:
20 |         # Target coverage for new code or changes
21 |         target: 80%
22 |         threshold: 5%
23 | 
24 | ignore:
25 |   # Deployment and configuration files
26 |   - "deploy/**/*"
27 |   - "scripts/**/*"
28 |   # Test files should not count toward coverage
29 |   - "tests/**/*"
30 |   # Setup and initialization files
31 |   - "setup.py"
32 |   - "aws_mcp_server/__main__.py"
33 |   - "aws_mcp_server/__init__.py"
34 |   # Documentation files
35 |   - "docs/**/*"
36 |   - "*.md"
37 |   # Version generated file
38 |   - "aws_mcp_server/_version.py"
39 | 
40 | comment:
41 |   layout: "reach, diff, flags, files"
42 |   behavior: default
43 |   require_changes: false
44 |   require_base: no
45 |   require_head: yes
```

--------------------------------------------------------------------------------
/deploy/docker/docker-compose.yml:
--------------------------------------------------------------------------------

```yaml
 1 | services:
 2 |   aws-mcp-server:
 3 |     # Use either local build or official image from GitHub Packages
 4 |     build:
 5 |       context: ../../
 6 |       dockerfile: ./deploy/docker/Dockerfile
 7 |     # Alternatively, use the pre-built multi-arch image
 8 |     # image: ghcr.io/alexei-led/aws-mcp-server:latest
 9 |     ports:
10 |       - "8000:8000"
11 |     volumes:
12 |       - ~/.aws://home/appuser/.aws:ro # Mount AWS credentials as read-only
13 |     environment:
14 |       - AWS_PROFILE=default # Specify default AWS profile
15 |       - AWS_MCP_TIMEOUT=300 # Default timeout in seconds (5 minutes)
16 |       - AWS_MCP_TRANSPORT=stdio # Transport protocol ("stdio" or "sse")
17 |       # - AWS_MCP_MAX_OUTPUT=100000  # Uncomment to set max output size
18 |     restart: unless-stopped
19 | # To build multi-architecture images:
20 | # 1. Set up Docker buildx: docker buildx create --name mybuilder --use
21 | # 2. Build and push the multi-arch image:
22 | #    docker buildx build --platform linux/amd64,linux/arm64 -t yourrepo/aws-mcp-server:latest --push .
23 | 
```

--------------------------------------------------------------------------------
/src/aws_mcp_server/__main__.py:
--------------------------------------------------------------------------------

```python
 1 | """Main entry point for the AWS MCP Server.
 2 | 
 3 | This module provides the entry point for running the AWS MCP Server.
 4 | FastMCP handles the command-line arguments and server configuration.
 5 | """
 6 | 
 7 | import logging
 8 | import signal
 9 | import sys
10 | 
11 | from aws_mcp_server.server import logger, mcp
12 | 
13 | # Configure root logger
14 | logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler(sys.stderr)])
15 | 
16 | 
17 | def handle_interrupt(signum, frame):
18 |     """Handle keyboard interrupt (Ctrl+C) gracefully."""
19 |     logger.info(f"Received signal {signum}, shutting down gracefully...")
20 |     sys.exit(0)
21 | 
22 | 
23 | # Using FastMCP's built-in CLI handling
24 | if __name__ == "__main__":
25 |     # Set up signal handler for graceful shutdown
26 |     signal.signal(signal.SIGINT, handle_interrupt)
27 |     signal.signal(signal.SIGTERM, handle_interrupt)
28 | 
29 |     try:
30 |         # Use configured transport protocol
31 |         from aws_mcp_server.config import TRANSPORT
32 | 
33 |         # Validate transport protocol
34 |         if TRANSPORT not in ("stdio", "sse"):
35 |             logger.error(f"Invalid transport protocol: {TRANSPORT}. Must be 'stdio' or 'sse'")
36 |             sys.exit(1)
37 | 
38 |         # Run with the specified transport protocol
39 |         logger.info(f"Starting server with transport protocol: {TRANSPORT}")
40 |         mcp.run(transport=TRANSPORT)
41 |     except KeyboardInterrupt:
42 |         logger.info("Keyboard interrupt received. Shutting down gracefully...")
43 |         sys.exit(0)
44 | 
```

--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------

```yaml
 1 | # Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
 2 | 
 3 | startCommand:
 4 |   type: stdio
 5 |   configSchema:
 6 |     # JSON Schema defining the configuration options for the MCP.
 7 |     type: object
 8 |     properties:
 9 |       awsMcpTimeout:
10 |         type: number
11 |         default: 300
12 |         description: Command execution timeout in seconds.
13 |       awsMcpMaxOutput:
14 |         type: number
15 |         default: 100000
16 |         description: Maximum output size in characters.
17 |       awsMcpTransport:
18 |         type: string
19 |         default: stdio
20 |         description: Transport protocol to use ('stdio' or 'sse').
21 |       awsProfile:
22 |         type: string
23 |         default: default
24 |         description: AWS profile to use.
25 |       awsRegion:
26 |         type: string
27 |         default: us-east-1
28 |         description: AWS region to use.
29 |   commandFunction:
30 |     # A JS function that produces the CLI command based on the given config to start the MCP on stdio.
31 |     |-
32 |     (config) => ({
33 |       command: 'python',
34 |       args: ['-m', 'aws_mcp_server'],
35 |       env: {
36 |         AWS_MCP_TIMEOUT: String(config.awsMcpTimeout || 300),
37 |         AWS_MCP_MAX_OUTPUT: String(config.awsMcpMaxOutput || 100000),
38 |         AWS_MCP_TRANSPORT: config.awsMcpTransport || 'stdio',
39 |         AWS_PROFILE: config.awsProfile || 'default',
40 |         AWS_REGION: config.awsRegion || 'us-east-1'
41 |       }
42 |     })
43 |   exampleConfig:
44 |     awsMcpTimeout: 300
45 |     awsMcpMaxOutput: 100000
46 |     awsMcpTransport: stdio
47 |     awsProfile: default
48 |     awsRegion: us-east-1
49 | 
50 | build:
51 |   dockerfile: deploy/docker/Dockerfile
52 |   dockerBuildPath: .
```

--------------------------------------------------------------------------------
/docs/VERSION.md:
--------------------------------------------------------------------------------

```markdown
 1 | # Version Management with setuptools_scm
 2 | 
 3 | This project uses [setuptools_scm](https://setuptools-scm.readthedocs.io/) to automatically determine version numbers from Git tags.
 4 | 
 5 | ## How it works
 6 | 
 7 | 1. The version is automatically determined from your git tags
 8 | 2. In development environments, the version is dynamically determined
 9 | 3. For Docker builds and CI, the version is passed as a build argument
10 | 
11 | ## Version Format
12 | 
13 | - Release: When on a tag (e.g., `1.2.3`), the version is exactly that tag
14 | - Development: When between tags, the version is `<last-tag>.post<n>+g<commit-hash>`
15 |   - Example: `1.2.3.post10+gb697684`
16 | 
17 | ## Local Development
18 | 
19 | The version is automatically determined whenever you:
20 | 
21 | ```bash
22 | # Install the package
23 | pip install -e .
24 | 
25 | # Run the version-file generator
26 | make version-file
27 | 
28 | # Check the current version
29 | python -m setuptools_scm
30 | ```
31 | 
32 | ## Importing Version in Code
33 | 
34 | ```python
35 | # Preferred method - via Python metadata
36 | from importlib.metadata import version
37 | __version__ = version("aws-mcp-server")
38 | 
39 | # Alternative - if using version file
40 | from aws_mcp_server._version import version, __version__
41 | ```
42 | 
43 | ## Docker and CI
44 | 
45 | For Docker builds, the version is:
46 | 
47 | 1. Determined by setuptools_scm
48 | 2. Passed to Docker as a build argument
49 | 3. Used in the image's labels and metadata
50 | 
51 | ## Creating Releases
52 | 
53 | To create a new release:
54 | 
55 | 1. Create and push a tag that follows semantic versioning:
56 |    ```bash
57 |    git tag -a 1.2.3 -m "Release 1.2.3"
58 |    git push origin 1.2.3
59 |    ```
60 | 
61 | 2. The CI pipeline will:
62 |    - Use setuptools_scm to get the version
63 |    - Build the Docker image with proper tags
64 |    - Push the release to registries
65 | 
66 | ## Usage Notes
67 | 
68 | - The `_version.py` file is automatically generated and ignored by git
69 | - Always include the patch version in tags (e.g., use `1.2.3` instead of `1.2`)
70 | - For the Docker image, the `+` in versions is replaced with `-` for compatibility
```

--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------

```toml
 1 | [build-system]
 2 | requires = ["setuptools>=61.0", "setuptools_scm>=8.0.0"]
 3 | build-backend = "setuptools.build_meta"
 4 | 
 5 | [project]
 6 | name = "aws-mcp-server"
 7 | dynamic = ["version"]
 8 | description = "AWS Model Context Protocol Server"
 9 | readme = "README.md"
10 | requires-python = ">=3.13"
11 | license = { text = "MIT" }
12 | authors = [{ name = "Alexei Ledenev" }]
13 | dependencies = [
14 |     "fastmcp>=0.4.1",
15 |     "mcp>=1.0.0",
16 |     "boto3>=1.34.0",
17 |     "pyyaml>=6.0.0"
18 | ]
19 | 
20 | [project.optional-dependencies]
21 | dev = [
22 |     "pytest>=7.0.0",
23 |     "pytest-cov>=4.0.0",
24 |     "pytest-asyncio>=0.23.0",
25 |     "ruff>=0.2.0",
26 |     "moto>=4.0.0",
27 |     "setuptools_scm>=7.0.0",
28 | ]
29 | # Production dependencies, optimized for Docker
30 | prod = [
31 |     "fastmcp>=0.4.1",
32 |     "mcp>=1.0.0",
33 |     "boto3>=1.34.0",
34 |     "pyyaml>=6.0.0",
35 | ]
36 | 
37 | [tool.setuptools]
38 | packages = ["aws_mcp_server"]
39 | package-dir = { "" = "src" }
40 | 
41 | [tool.ruff]
42 | line-length = 160
43 | target-version = "py313"
44 | exclude = ["src/aws_mcp_server/_version.py"]
45 | 
46 | [tool.ruff.lint]
47 | select = ["E", "F", "I", "B"]
48 | 
49 | [tool.ruff.format]
50 | quote-style = "double"
51 | indent-style = "space"
52 | line-ending = "auto"
53 | 
54 | [tool.ruff.lint.isort]
55 | known-first-party = ["aws_mcp_server"]
56 | 
57 | # Using VSCode + Pylance static typing instead of mypy
58 | 
59 | [tool.pytest.ini_options]
60 | testpaths = ["tests"]
61 | python_files = "test_*.py"
62 | markers = [
63 |     "integration: marks tests that require AWS CLI and AWS credentials",
64 |     "asyncio: mark test as requiring asyncio",
65 | ]
66 | asyncio_mode = "strict"
67 | asyncio_default_fixture_loop_scope = "function"
68 | filterwarnings = [
69 |     "ignore::RuntimeWarning:unittest.mock:",
70 |     "ignore::RuntimeWarning:weakref:"
71 | ]
72 | 
73 | [tool.coverage.run]
74 | source = ["src/aws_mcp_server"]
75 | omit = [
76 |     "*/tests/*",
77 |     "*/setup.py",
78 |     "*/conftest.py",
79 |     "src/aws_mcp_server/__main__.py",
80 | ]
81 | 
82 | [tool.coverage.report]
83 | exclude_lines = [
84 |     "pragma: no cover",
85 |     "def __repr__",
86 |     "if self.debug",
87 |     "raise NotImplementedError",
88 |     "if __name__ == .__main__.:",
89 |     "pass",
90 |     "raise ImportError",
91 | ]
92 | 
93 | [tool.setuptools_scm]
94 | fallback_version="0.0.0-dev0"
```

--------------------------------------------------------------------------------
/tests/unit/test_init.py:
--------------------------------------------------------------------------------

```python
 1 | """Tests for the package initialization module."""
 2 | 
 3 | import unittest
 4 | from importlib import reload
 5 | from unittest.mock import patch
 6 | 
 7 | 
 8 | class TestInitModule(unittest.TestCase):
 9 |     """Tests for the __init__ module."""
10 | 
11 |     def test_version_from_package(self):
12 |         """Test __version__ is set from package metadata."""
13 |         with patch("importlib.metadata.version", return_value="1.2.3"):
14 |             # Import the module fresh to apply the patch
15 |             import aws_mcp_server
16 | 
17 |             # Reload to apply our patch
18 |             reload(aws_mcp_server)
19 | 
20 |             # Check that __version__ is set correctly
21 |             self.assertEqual(aws_mcp_server.__version__, "1.2.3")
22 | 
23 |     def test_version_fallback_on_package_not_found(self):
24 |         """Test handling of PackageNotFoundError."""
25 |         from importlib.metadata import PackageNotFoundError
26 | 
27 |         # Looking at the actual implementation, when PackageNotFoundError is raised,
28 |         # it just uses 'pass', so the attribute __version__ may or may not be set.
29 |         # If it was previously set (which is likely), it will retain its previous value.
30 |         with patch("importlib.metadata.version", side_effect=PackageNotFoundError):
31 |             # Create a fresh module
32 |             import sys
33 | 
34 |             if "aws_mcp_server" in sys.modules:
35 |                 del sys.modules["aws_mcp_server"]
36 | 
37 |             # Import the module fresh with our patch
38 |             import aws_mcp_server
39 | 
40 |             # In this case, the __version__ may not even be set
41 |             # We're just testing that the code doesn't crash with PackageNotFoundError
42 |             # Our test should pass regardless of whether __version__ is set
43 |             # The important part is that the exception is handled
44 |             try:
45 |                 # This could raise AttributeError
46 |                 _ = aws_mcp_server.__version__
47 |                 # If we get here, it's set to something - hard to assert exactly what
48 |                 # Just ensure no exception was thrown
49 |                 self.assertTrue(True)
50 |             except AttributeError:
51 |                 # If AttributeError is raised, that's also fine - the attribute doesn't exist
52 |                 self.assertTrue(True)
53 | 
54 | 
55 | if __name__ == "__main__":
56 |     unittest.main()
57 | 
```

--------------------------------------------------------------------------------
/tests/test_bucket_creation.py:
--------------------------------------------------------------------------------

```python
 1 | """Test for creating and managing S3 buckets directly."""
 2 | 
 3 | import asyncio
 4 | import os
 5 | import time
 6 | import uuid
 7 | 
 8 | import pytest
 9 | 
10 | from aws_mcp_server.config import AWS_REGION
11 | from aws_mcp_server.server import aws_cli_pipeline
12 | 
13 | 
14 | @pytest.mark.integration
15 | @pytest.mark.asyncio
16 | async def test_create_and_delete_s3_bucket():
17 |     """Test creating and deleting an S3 bucket using AWS MCP server."""
18 |     # Get region from environment or use default
19 |     region = os.environ.get("AWS_TEST_REGION", AWS_REGION)
20 |     print(f"Using AWS region: {region}")
21 | 
22 |     # Generate a unique bucket name
23 |     timestamp = int(time.time())
24 |     random_id = str(uuid.uuid4())[:8]
25 |     bucket_name = f"aws-mcp-test-{timestamp}-{random_id}"
26 | 
27 |     try:
28 |         # Create the bucket
29 |         create_cmd = f"aws s3 mb s3://{bucket_name} --region {region}"
30 |         result = await aws_cli_pipeline(command=create_cmd, timeout=None, ctx=None)
31 | 
32 |         # Check if bucket was created successfully
33 |         assert result["status"] == "success", f"Failed to create bucket: {result['output']}"
34 | 
35 |         # Wait for bucket to be fully available
36 |         await asyncio.sleep(3)
37 | 
38 |         # List buckets to verify it exists
39 |         list_result = await aws_cli_pipeline(command="aws s3 ls", timeout=None, ctx=None)
40 |         assert bucket_name in list_result["output"], "Bucket not found in bucket list"
41 | 
42 |         # Try to create a test file
43 |         test_content = "Test content"
44 |         with open("test_file.txt", "w") as f:
45 |             f.write(test_content)
46 | 
47 |         # Upload the file
48 |         upload_result = await aws_cli_pipeline(command=f"aws s3 cp test_file.txt s3://{bucket_name}/test_file.txt --region {region}", timeout=None, ctx=None)
49 |         assert upload_result["status"] == "success", f"Failed to upload file: {upload_result['output']}"
50 | 
51 |         # List bucket contents
52 |         list_files_result = await aws_cli_pipeline(command=f"aws s3 ls s3://{bucket_name}/ --region {region}", timeout=None, ctx=None)
53 |         assert "test_file.txt" in list_files_result["output"], "Uploaded file not found in bucket"
54 | 
55 |     finally:
56 |         # Clean up
57 |         # Remove test file
58 |         if os.path.exists("test_file.txt"):
59 |             os.remove("test_file.txt")
60 | 
61 |         # Delete all objects in the bucket
62 |         await aws_cli_pipeline(command=f"aws s3 rm s3://{bucket_name} --recursive --region {region}", timeout=None, ctx=None)
63 | 
64 |         # Delete the bucket
65 |         delete_result = await aws_cli_pipeline(command=f"aws s3 rb s3://{bucket_name} --region {region}", timeout=None, ctx=None)
66 |         assert delete_result["status"] == "success", f"Failed to delete bucket: {delete_result['output']}"
67 | 
```

--------------------------------------------------------------------------------
/tests/unit/test_main.py:
--------------------------------------------------------------------------------

```python
 1 | """Tests for the main entry point of the AWS MCP Server."""
 2 | 
 3 | from unittest.mock import MagicMock, patch
 4 | 
 5 | import pytest
 6 | 
 7 | # Import handle_interrupt function for direct testing
 8 | from aws_mcp_server.__main__ import handle_interrupt
 9 | 
10 | 
11 | def test_handle_interrupt():
12 |     """Test the handle_interrupt function."""
13 |     with patch("sys.exit") as mock_exit:
14 |         # Call the function with mock signal and frame
15 |         handle_interrupt(MagicMock(), MagicMock())
16 |         # Verify sys.exit was called with 0
17 |         mock_exit.assert_called_once_with(0)
18 | 
19 | 
20 | @pytest.mark.skip(reason="Cannot reload main module during testing")
21 | def test_main_with_valid_transport():
22 |     """Test main module with valid transport setting."""
23 |     with patch("aws_mcp_server.__main__.TRANSPORT", "stdio"):
24 |         with patch("aws_mcp_server.__main__.mcp.run") as mock_run:
25 |             # We can't easily test the full __main__ module execution
26 |             from aws_mcp_server.__main__ import mcp
27 | 
28 |             # Instead, we'll test the specific function we modified
29 |             with patch("aws_mcp_server.__main__.logger") as mock_logger:
30 |                 # Import the function to ensure proper validation
31 |                 from aws_mcp_server.__main__ import TRANSPORT
32 | 
33 |                 # Call the relevant function directly
34 |                 mcp.run(transport=TRANSPORT)
35 | 
36 |                 # Check that mcp.run was called with the correct transport
37 |                 mock_run.assert_called_once_with(transport="stdio")
38 |                 # Verify logger was called
39 |                 mock_logger.info.assert_any_call("Starting server with transport protocol: stdio")
40 | 
41 | 
42 | def test_main_transport_validation():
43 |     """Test transport protocol validation."""
44 |     with patch("aws_mcp_server.config.TRANSPORT", "invalid"):
45 |         from aws_mcp_server.config import TRANSPORT
46 | 
47 |         # Test the main function's validation logic
48 |         with patch("aws_mcp_server.server.mcp.run") as mock_run:
49 |             with patch("sys.exit") as mock_exit:
50 |                 with patch("aws_mcp_server.__main__.logger") as mock_logger:
51 |                     # Execute the validation logic directly
52 |                     if TRANSPORT not in ("stdio", "sse"):
53 |                         mock_logger.error(f"Invalid transport protocol: {TRANSPORT}. Must be 'stdio' or 'sse'")
54 |                         mock_exit(1)
55 |                     else:
56 |                         mock_run(transport=TRANSPORT)
57 | 
58 |                     # Check that error was logged with invalid transport
59 |                     mock_logger.error.assert_called_once_with("Invalid transport protocol: invalid. Must be 'stdio' or 'sse'")
60 |                     # Check that exit was called
61 |                     mock_exit.assert_called_once_with(1)
62 |                     # Check that mcp.run was not called
63 |                     mock_run.assert_not_called()
64 | 
```

--------------------------------------------------------------------------------
/.github/workflows/ci.yml:
--------------------------------------------------------------------------------

```yaml
  1 | name: PR Validation
  2 | 
  3 | on:
  4 |   pull_request:
  5 |     paths-ignore:
  6 |       - 'deploy/**'
  7 |       - '*.md'
  8 | 
  9 | jobs:
 10 |   test:
 11 |     runs-on: ubuntu-latest
 12 |     if: "!contains(github.event.head_commit.message, '[ci skip]') && !contains(github.event.head_commit.message, '[skip ci]')"
 13 |     strategy:
 14 |       matrix:
 15 |         python-version: ["3.13"]
 16 | 
 17 |     steps:
 18 |       - uses: actions/checkout@v4
 19 | 
 20 |       - name: Set up Python ${{ matrix.python-version }}
 21 |         uses: actions/setup-python@v5
 22 |         with:
 23 |           python-version: ${{ matrix.python-version }}
 24 |           cache: "pip"
 25 | 
 26 |       - name: Install uv
 27 |         run: |
 28 |           # Install uv using the official installation method
 29 |           curl -LsSf https://astral.sh/uv/install.sh | sh
 30 | 
 31 |           # Add uv to PATH
 32 |           echo "$HOME/.cargo/bin" >> $GITHUB_PATH
 33 | 
 34 |       - name: Install dependencies using uv
 35 |         run: |
 36 |           # Install dependencies using uv with the lock file and the --system flag
 37 |           uv pip install --system -e ".[dev]"
 38 | 
 39 |       - name: Lint
 40 |         run: make lint
 41 |         continue-on-error: true  # Display errors but don't fail build for lint warnings
 42 | 
 43 |       - name: Test
 44 |         run: make test
 45 | 
 46 |       - name: Upload coverage to Codecov
 47 |         uses: codecov/codecov-action@v4
 48 |         with:
 49 |           token: ${{ secrets.CODECOV_TOKEN }}
 50 |           file: ./coverage.xml
 51 |           fail_ci_if_error: false
 52 |           verbose: true
 53 | 
 54 |   build:
 55 |     runs-on: ubuntu-latest
 56 |     needs: test
 57 |     steps:
 58 |       - uses: actions/checkout@v4
 59 | 
 60 |       - name: Set up Docker Buildx
 61 |         uses: docker/setup-buildx-action@v3
 62 | 
 63 |       - name: Get current date
 64 |         id: date
 65 |         run: echo "date=$(date -u +'%Y-%m-%dT%H:%M:%SZ')" >> $GITHUB_OUTPUT
 66 | 
 67 |       - name: Install setuptools_scm
 68 |         run: pip install setuptools_scm
 69 |         
 70 |       - name: Generate version file and get version info
 71 |         id: version
 72 |         run: |
 73 |           # Generate version file automatically
 74 |           python -m setuptools_scm
 75 |           
 76 |           # Get the raw version from setuptools_scm
 77 |           VERSION=$(python -m setuptools_scm)
 78 |           
 79 |           # Make version Docker-compatible (replace + with -)
 80 |           DOCKER_VERSION=$(echo "$VERSION" | tr '+' '-')
 81 |           
 82 |           # Update the version in pyproject.toml
 83 |           sed -i "s|fallback_version=\"0.0.0-dev0\"|fallback_version=\"${VERSION}\"|g" pyproject.toml
 84 |           
 85 |           echo "version=$DOCKER_VERSION" >> $GITHUB_OUTPUT
 86 | 
 87 |       - name: Build Docker image
 88 |         uses: docker/build-push-action@v5
 89 |         with:
 90 |           context: .
 91 |           file: ./deploy/docker/Dockerfile
 92 |           push: false
 93 |           tags: aws-mcp-server:${{ steps.version.outputs.version }}
 94 |           platforms: linux/amd64
 95 |           build-args: |
 96 |             BUILD_DATE=${{ steps.date.outputs.date }}
 97 |             VERSION=${{ steps.version.outputs.version }}
 98 |           cache-from: type=gha
 99 |           cache-to: type=gha,mode=max
100 | 
```

--------------------------------------------------------------------------------
/deploy/docker/Dockerfile:
--------------------------------------------------------------------------------

```dockerfile
  1 | # Multi-stage build with platform-specific configuration
  2 | ARG PYTHON_VERSION=3.13-slim
  3 | ARG VERSION
  4 | 
  5 | # =========== BUILDER STAGE ===========
  6 | FROM --platform=${TARGETPLATFORM} python:${PYTHON_VERSION} AS builder
  7 | 
  8 | # Install build dependencies
  9 | RUN apt-get update && apt-get install -y --no-install-recommends \
 10 |     build-essential \
 11 |     && apt-get clean \
 12 |     && rm -rf /var/lib/apt/lists/*
 13 | 
 14 | # Set up working directory
 15 | WORKDIR /build
 16 | 
 17 | # Copy package definition files
 18 | COPY pyproject.toml README.md LICENSE ./
 19 | COPY src/ ./src/
 20 | 
 21 | RUN cat pyproject.toml
 22 | 
 23 | # Install package and dependencies with pip wheel
 24 | RUN pip install --no-cache-dir wheel && \
 25 |     pip wheel --no-cache-dir --wheel-dir=/wheels -e .
 26 | 
 27 | # =========== FINAL STAGE ===========
 28 | FROM --platform=${TARGETPLATFORM} python:${PYTHON_VERSION}
 29 | 
 30 | # Set target architecture argument
 31 | ARG TARGETPLATFORM
 32 | ARG TARGETARCH
 33 | ARG BUILD_DATE
 34 | ARG VERSION
 35 | 
 36 | # Add metadata
 37 | LABEL maintainer="alexei-led" \
 38 |       description="AWS Multi-Command Proxy Server" \
 39 |       org.opencontainers.image.source="https://github.com/alexei-led/aws-mcp-server" \
 40 |       org.opencontainers.image.version="${VERSION}" \
 41 |       org.opencontainers.image.created="${BUILD_DATE}"
 42 | 
 43 | # Step 1: Install system packages - keeping all original packages
 44 | RUN apt-get update && apt-get install -y --no-install-recommends \
 45 |     unzip \
 46 |     curl \
 47 |     wget \
 48 |     less \
 49 |     groff \
 50 |     jq \
 51 |     gnupg \
 52 |     tar \
 53 |     gzip \
 54 |     zip \
 55 |     vim \
 56 |     net-tools \
 57 |     dnsutils \
 58 |     openssh-client \
 59 |     grep \
 60 |     sed \
 61 |     gawk \
 62 |     findutils \
 63 |     && apt-get clean \
 64 |     && rm -rf /var/lib/apt/lists/*
 65 | 
 66 | # Step 2: Install AWS CLI based on architecture
 67 | RUN if [ "${TARGETARCH}" = "arm64" ]; then \
 68 |         curl -sSL "https://awscli.amazonaws.com/awscli-exe-linux-aarch64.zip" -o "awscliv2.zip"; \
 69 |     else \
 70 |         curl -sSL "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"; \
 71 |     fi \
 72 |     && unzip -q awscliv2.zip \
 73 |     && ./aws/install \
 74 |     && rm -rf awscliv2.zip aws
 75 | 
 76 | # Step 3: Install Session Manager plugin (only for x86_64 due to compatibility issues on ARM)
 77 | RUN if [ "${TARGETARCH}" = "amd64" ]; then \
 78 |         curl -sSL "https://s3.amazonaws.com/session-manager-downloads/plugin/latest/ubuntu_64bit/session-manager-plugin.deb" -o "session-manager-plugin.deb" \
 79 |         && dpkg -i session-manager-plugin.deb 2>/dev/null || apt-get -f install -y \
 80 |         && rm session-manager-plugin.deb; \
 81 |     else \
 82 |         echo "Skipping Session Manager plugin installation for ${TARGETARCH} architecture"; \
 83 |     fi
 84 | 
 85 | # Set up application directory, user, and permissions
 86 | RUN useradd -m -s /bin/bash -u 10001 appuser \
 87 |     && mkdir -p /app/logs && chmod 777 /app/logs \
 88 |     && mkdir -p /home/appuser/.aws && chmod 700 /home/appuser/.aws
 89 | 
 90 | WORKDIR /app
 91 | 
 92 | # Copy application code
 93 | COPY pyproject.toml README.md LICENSE ./
 94 | COPY src/ ./src/
 95 | 
 96 | # Copy wheels from builder and install
 97 | COPY --from=builder /wheels /wheels
 98 | RUN pip install --no-cache-dir --no-index --find-links=/wheels aws-mcp-server && \
 99 |     rm -rf /wheels
100 | 
101 | # Set ownership after all files have been copied - avoid .aws directory
102 | RUN chown -R appuser:appuser /app
103 | 
104 | # Switch to non-root user
105 | USER appuser
106 | 
107 | # Set all environment variables in one layer
108 | ENV HOME="/home/appuser" \
109 |     PATH="/usr/local/bin:/usr/local/aws/v2/bin:${PATH}" \
110 |     PYTHONUNBUFFERED=1 \
111 |     AWS_MCP_TRANSPORT=stdio
112 | 
113 | # Expose the service port
114 | EXPOSE 8000
115 | 
116 | # Set command to run the server
117 | ENTRYPOINT ["python", "-m", "aws_mcp_server"]
```

--------------------------------------------------------------------------------
/tests/test_aws_setup.py:
--------------------------------------------------------------------------------

```python
 1 | """Test file to verify AWS integration setup works correctly."""
 2 | 
 3 | import asyncio
 4 | import os
 5 | import subprocess
 6 | import time
 7 | import uuid
 8 | from unittest.mock import AsyncMock, patch
 9 | 
10 | import pytest
11 | 
12 | from aws_mcp_server.server import aws_cli_pipeline
13 | 
14 | 
15 | def test_aws_cli_installed():
16 |     """Test that AWS CLI is installed."""
17 |     result = subprocess.run(["aws", "--version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
18 |     assert result.returncode == 0, "AWS CLI is not installed or not in PATH"
19 | 
20 | 
21 | @pytest.mark.integration
22 | def test_aws_credentials_exist():
23 |     """Test that AWS credentials exist.
24 | 
25 |     This test is marked as integration because it requires AWS credentials.
26 |     """
27 |     result = subprocess.run(["aws", "sts", "get-caller-identity"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
28 |     assert result.returncode == 0, f"AWS credentials check failed: {result.stderr.decode('utf-8')}"
29 | 
30 | 
31 | @pytest.mark.asyncio
32 | @pytest.mark.integration
33 | async def test_aws_execute_command():
34 |     """Test that we can execute a basic AWS command.
35 | 
36 |     This test is marked as integration because it requires AWS credentials.
37 |     """
38 |     # Test a simple S3 bucket listing command
39 |     result = await aws_cli_pipeline(command="aws s3 ls", timeout=None, ctx=None)
40 | 
41 |     # Verify the result
42 |     assert isinstance(result, dict)
43 |     assert "status" in result
44 |     assert result["status"] == "success", f"Command failed: {result.get('output', '')}"
45 | 
46 | 
47 | @pytest.mark.asyncio
48 | @pytest.mark.integration
49 | async def test_aws_bucket_creation():
50 |     """Test that we can create and delete a bucket.
51 | 
52 |     This test is marked as integration because it requires AWS credentials.
53 |     """
54 |     # Generate a bucket name
55 |     timestamp = int(time.time())
56 |     random_id = str(uuid.uuid4())[:8]
57 |     bucket_name = f"aws-mcp-test-{timestamp}-{random_id}"
58 | 
59 |     # Get region from environment or use default
60 |     region = os.environ.get("AWS_TEST_REGION", os.environ.get("AWS_REGION", "us-east-1"))
61 | 
62 |     try:
63 |         # Create bucket with region specification
64 |         create_result = await aws_cli_pipeline(command=f"aws s3 mb s3://{bucket_name} --region {region}", timeout=None, ctx=None)
65 |         assert create_result["status"] == "success", f"Failed to create bucket: {create_result['output']}"
66 | 
67 |         # Verify bucket exists
68 |         await asyncio.sleep(3)  # Wait for bucket to be fully available
69 |         list_result = await aws_cli_pipeline(command="aws s3 ls", timeout=None, ctx=None)
70 |         assert bucket_name in list_result["output"], "Bucket was not found in bucket list"
71 | 
72 |     finally:
73 |         # Clean up - delete the bucket
74 |         await aws_cli_pipeline(command=f"aws s3 rb s3://{bucket_name} --region {region}", timeout=None, ctx=None)
75 | 
76 | 
77 | @pytest.mark.asyncio
78 | async def test_aws_command_mocked():
79 |     """Test executing an AWS command with mocked execution.
80 | 
81 |     This test is mocked so it doesn't require AWS credentials, suitable for CI.
82 |     """
83 |     # We need to patch the correct module path
84 |     with patch("aws_mcp_server.server.execute_aws_command", new_callable=AsyncMock) as mock_execute:
85 |         # Set up mock return value
86 |         mock_execute.return_value = {"status": "success", "output": "Mock bucket list output"}
87 | 
88 |         # Execute the command
89 |         result = await aws_cli_pipeline(command="aws s3 ls", timeout=None, ctx=None)
90 | 
91 |         # Verify the mock was called correctly
92 |         mock_execute.assert_called_once()
93 | 
94 |         # Check the results
95 |         assert result["status"] == "success"
96 |         assert "Mock bucket list output" in result["output"]
97 | 
```

--------------------------------------------------------------------------------
/src/aws_mcp_server/config.py:
--------------------------------------------------------------------------------

```python
 1 | """Configuration settings for the AWS MCP Server.
 2 | 
 3 | This module contains configuration settings for the AWS MCP Server.
 4 | 
 5 | Environment variables:
 6 | - AWS_MCP_TIMEOUT: Custom timeout in seconds (default: 300)
 7 | - AWS_MCP_MAX_OUTPUT: Maximum output size in characters (default: 100000)
 8 | - AWS_MCP_TRANSPORT: Transport protocol to use ("stdio" or "sse", default: "stdio")
 9 | - AWS_PROFILE: AWS profile to use (default: "default")
10 | - AWS_REGION: AWS region to use (default: "us-east-1")
11 | - AWS_DEFAULT_REGION: Alternative to AWS_REGION (used if AWS_REGION not set)
12 | - AWS_MCP_SECURITY_MODE: Security mode for command validation (strict or permissive, default: strict)
13 | - AWS_MCP_SECURITY_CONFIG: Path to custom security configuration file
14 | """
15 | 
16 | import os
17 | from pathlib import Path
18 | 
19 | # Command execution settings
20 | DEFAULT_TIMEOUT = int(os.environ.get("AWS_MCP_TIMEOUT", "300"))
21 | MAX_OUTPUT_SIZE = int(os.environ.get("AWS_MCP_MAX_OUTPUT", "100000"))
22 | 
23 | # Transport protocol
24 | TRANSPORT = os.environ.get("AWS_MCP_TRANSPORT", "stdio")
25 | 
26 | # AWS CLI settings
27 | AWS_PROFILE = os.environ.get("AWS_PROFILE", "default")
28 | AWS_REGION = os.environ.get("AWS_REGION", os.environ.get("AWS_DEFAULT_REGION", "us-east-1"))
29 | 
30 | # Security settings
31 | SECURITY_MODE = os.environ.get("AWS_MCP_SECURITY_MODE", "strict")
32 | SECURITY_CONFIG_PATH = os.environ.get("AWS_MCP_SECURITY_CONFIG", "")
33 | 
34 | # Instructions displayed to client during initialization
35 | INSTRUCTIONS = """
36 | AWS MCP Server provides a comprehensive interface to the AWS CLI with best practices guidance.
37 | - Use the describe_command tool to get AWS CLI documentation
38 | - Use the execute_command tool to run AWS CLI commands
39 | - The execute_command tool supports Unix pipes (|) to filter or transform AWS CLI output:
40 |   Example: aws s3api list-buckets --query 'Buckets[*].Name' --output text | sort
41 | - Access AWS environment resources for context:
42 |   - aws://config/profiles: List available AWS profiles and active profile
43 |   - aws://config/regions: List available AWS regions and active region
44 |   - aws://config/regions/{region}: Get detailed information about a specific region 
45 |     including name, code, availability zones, geographic location, and available services
46 |   - aws://config/environment: Get current AWS environment details (profile, region, credentials)
47 |   - aws://config/account: Get current AWS account information (ID, alias, organization)
48 | - Use the built-in prompt templates for common AWS tasks following AWS Well-Architected Framework best practices:
49 | 
50 |   Essential Operations:
51 |   - create_resource: Create AWS resources with comprehensive security settings
52 |   - resource_inventory: Create detailed resource inventories across regions
53 |   - troubleshoot_service: Perform systematic service issue diagnostics
54 | 
55 |   Security & Compliance:
56 |   - security_audit: Perform comprehensive service security audits
57 |   - security_posture_assessment: Evaluate overall AWS security posture
58 |   - iam_policy_generator: Generate least-privilege IAM policies
59 |   - compliance_check: Verify compliance with regulatory standards
60 | 
61 |   Cost & Performance:
62 |   - cost_optimization: Find and implement cost optimization opportunities
63 |   - resource_cleanup: Safely clean up unused resources
64 |   - performance_tuning: Optimize performance for specific resources
65 | 
66 |   Infrastructure & Architecture:
67 |   - serverless_deployment: Deploy serverless applications with best practices
68 |   - container_orchestration: Set up container environments (ECS/EKS)
69 |   - vpc_network_design: Design and deploy secure VPC networking
70 |   - infrastructure_automation: Automate infrastructure management
71 |   - multi_account_governance: Implement secure multi-account strategies
72 | 
73 |   Reliability & Monitoring:
74 |   - service_monitoring: Configure comprehensive service monitoring
75 |   - disaster_recovery: Implement enterprise-grade DR solutions
76 | """
77 | 
78 | # Application paths
79 | BASE_DIR = Path(__file__).parent.parent.parent
80 | 
```

--------------------------------------------------------------------------------
/.github/workflows/release.yml:
--------------------------------------------------------------------------------

```yaml
  1 | name: Release
  2 | 
  3 | on:
  4 |   push:
  5 |     branches:
  6 |       - master
  7 |       - main
  8 |     tags:
  9 |       - '[0-9]+.[0-9]+.[0-9]+'
 10 |       - 'v[0-9]+.[0-9]+.[0-9]+'
 11 |     paths-ignore:
 12 |       - 'tests/**'
 13 |       - '*.md'
 14 | 
 15 | jobs:
 16 |   build-and-push:
 17 |     runs-on: ubuntu-latest
 18 |     if: "!contains(github.event.head_commit.message, '[ci skip]') && !contains(github.event.head_commit.message, '[skip ci]')"
 19 | 
 20 |     permissions:
 21 |       contents: read
 22 |       packages: write
 23 | 
 24 |     steps:
 25 |       - uses: actions/checkout@v4
 26 | 
 27 |       - name: Set up Python 3.13
 28 |         uses: actions/setup-python@v5
 29 |         with:
 30 |           python-version: "3.13"
 31 |           cache: "pip"
 32 | 
 33 |       - name: Install dependencies and run tests
 34 |         run: |
 35 |           python -m pip install -e ".[dev]"
 36 |           # Run linting and tests to verify before release
 37 |           make lint
 38 |           make test
 39 |           
 40 |       - name: Upload coverage to Codecov
 41 |         uses: codecov/codecov-action@v4
 42 |         with:
 43 |           token: ${{ secrets.CODECOV_TOKEN }}
 44 |           file: ./coverage.xml
 45 |           fail_ci_if_error: false
 46 |           verbose: true
 47 | 
 48 |       - name: Log in to GitHub Container Registry
 49 |         uses: docker/login-action@v3
 50 |         with:
 51 |           registry: ghcr.io
 52 |           username: ${{ github.actor }}
 53 |           password: ${{ secrets.GITHUB_TOKEN }}
 54 | 
 55 |       - name: Install setuptools_scm
 56 |         run: pip install setuptools_scm
 57 | 
 58 |       - name: Generate version file and get version information
 59 |         id: version
 60 |         run: |
 61 |           # Generate version file automatically 
 62 |           VERSION=$(python -m setuptools_scm)
 63 |           
 64 |           # Check if we're on a tag
 65 |           if [[ "${{ github.ref_type }}" == "tag" ]]; then
 66 |             echo "is_tag=true" >> $GITHUB_OUTPUT
 67 |             
 68 |             # Parse semver components for tagging
 69 |             VERSION_NO_V=$(echo "${{ github.ref_name }}" | sed 's/^v//')
 70 |             # overwrite VERSION with the tag name
 71 |             VERSION=${VERSION_NO_V}
 72 |             MAJOR=$(echo "${VERSION_NO_V}" | cut -d. -f1)
 73 |             MINOR=$(echo "${VERSION_NO_V}" | cut -d. -f2)
 74 |             PATCH=$(echo "${VERSION_NO_V}" | cut -d. -f3)
 75 |             
 76 |             echo "major=${MAJOR}" >> $GITHUB_OUTPUT
 77 |             echo "major_minor=${MAJOR}.${MINOR}" >> $GITHUB_OUTPUT
 78 |             echo "major_minor_patch=${VERSION_NO_V}" >> $GITHUB_OUTPUT
 79 |             echo "version=${VERSION_NO_V}" >> $GITHUB_OUTPUT
 80 |           else
 81 |             # For non-tag builds, use setuptools_scm
 82 |             VERSION=$(python -m setuptools_scm)
 83 |             # Make version Docker-compatible (replace + with -)
 84 |             DOCKER_VERSION=$(echo "$VERSION" | tr '+' '-')
 85 |             echo "is_tag=false" >> $GITHUB_OUTPUT
 86 |             echo "version=${DOCKER_VERSION}" >> $GITHUB_OUTPUT
 87 |           fi
 88 |           echo "build_date=$(date -u +'%Y-%m-%dT%H:%M:%SZ')" >> $GITHUB_OUTPUT
 89 |           
 90 |           # Update the version in pyproject.toml
 91 |           sed -i "s|fallback_version=\"0.0.0-dev0\"|fallback_version=\"${VERSION}\"|g" pyproject.toml
 92 | 
 93 |       - name: Extract metadata for Docker
 94 |         id: meta
 95 |         uses: docker/metadata-action@v5
 96 |         with:
 97 |           images: ghcr.io/${{ github.repository }}
 98 |           tags: |
 99 |             # For tags: exact semver from the tag name
100 |             type=raw,value=${{ steps.version.outputs.major_minor_patch }},enable=${{ steps.version.outputs.is_tag == 'true' }}
101 |             type=raw,value=${{ steps.version.outputs.major_minor }},enable=${{ steps.version.outputs.is_tag == 'true' }}
102 |             type=raw,value=${{ steps.version.outputs.major }},enable=${{ steps.version.outputs.is_tag == 'true' }}
103 |             type=raw,value=latest,enable=${{ steps.version.outputs.is_tag == 'true' }}
104 |             # Git SHA for both tag and non-tag builds
105 |             type=sha,format=short
106 |             # For main branch: dev tag
107 |             type=raw,value=dev,enable=${{ github.ref == format('refs/heads/{0}', 'main') }}
108 | 
109 |       - name: Set up Docker Buildx
110 |         uses: docker/setup-buildx-action@v3
111 | 
112 |       - name: Build and push multi-architecture Docker image
113 |         uses: docker/build-push-action@v6
114 |         with:
115 |           context: .
116 |           file: ./deploy/docker/Dockerfile
117 |           push: true
118 |           platforms: linux/amd64,linux/arm64
119 |           tags: ${{ steps.meta.outputs.tags }}
120 |           labels: ${{ steps.meta.outputs.labels }}
121 |           build-args: |
122 |             BUILD_DATE=${{ steps.version.outputs.build_date }}
123 |             VERSION=${{ steps.version.outputs.version }}
124 |           cache-from: type=gha
125 |           cache-to: type=gha,mode=max
126 | 
```

--------------------------------------------------------------------------------
/tests/unit/test_prompts.py:
--------------------------------------------------------------------------------

```python
  1 | """Unit tests for AWS MCP Server prompts.
  2 | 
  3 | Tests the prompt templates functionality in the AWS MCP Server.
  4 | """
  5 | 
  6 | from unittest.mock import MagicMock
  7 | 
  8 | import pytest
  9 | 
 10 | from aws_mcp_server.prompts import register_prompts
 11 | 
 12 | 
 13 | @pytest.fixture
 14 | def prompt_functions():
 15 |     """Fixture that returns a dictionary of prompt functions.
 16 | 
 17 |     This fixture captures all prompt functions registered with the MCP instance.
 18 |     """
 19 |     captured_functions = {}
 20 | 
 21 |     # Create a special mock decorator that captures the functions
 22 |     def mock_prompt_decorator(*args, **kwargs):
 23 |         def decorator(func):
 24 |             captured_functions[func.__name__] = func
 25 |             return func
 26 | 
 27 |         return decorator
 28 | 
 29 |     mock_mcp = MagicMock()
 30 |     mock_mcp.prompt = mock_prompt_decorator
 31 | 
 32 |     # Register prompts with our special mock
 33 |     register_prompts(mock_mcp)
 34 | 
 35 |     return captured_functions
 36 | 
 37 | 
 38 | def test_prompt_registration(prompt_functions):
 39 |     """Test that prompts are registered correctly."""
 40 |     # All expected prompt names
 41 |     expected_prompt_names = [
 42 |         "create_resource",
 43 |         "security_audit",
 44 |         "cost_optimization",
 45 |         "resource_inventory",
 46 |         "troubleshoot_service",
 47 |         "iam_policy_generator",
 48 |         "service_monitoring",
 49 |         "disaster_recovery",
 50 |         "compliance_check",
 51 |         "resource_cleanup",
 52 |         "serverless_deployment",
 53 |         "container_orchestration",
 54 |         "vpc_network_design",
 55 |         "infrastructure_automation",
 56 |         "security_posture_assessment",
 57 |         "performance_tuning",
 58 |         "multi_account_governance",
 59 |     ]
 60 | 
 61 |     # Check that we captured the expected number of functions
 62 |     assert len(prompt_functions) == len(expected_prompt_names), f"Expected {len(expected_prompt_names)} prompts, got {len(prompt_functions)}"
 63 | 
 64 |     # Check that all expected prompts are registered
 65 |     for prompt_name in expected_prompt_names:
 66 |         assert prompt_name in prompt_functions, f"Expected prompt '{prompt_name}' not found"
 67 | 
 68 | 
 69 | @pytest.mark.parametrize(
 70 |     "prompt_name,args,expected_content",
 71 |     [
 72 |         # Original prompts
 73 |         ("create_resource", {"resource_type": "s3-bucket", "resource_name": "my-test-bucket"}, ["s3-bucket", "my-test-bucket", "security", "best practices"]),
 74 |         ("security_audit", {"service": "s3"}, ["s3", "security audit", "public access"]),
 75 |         ("cost_optimization", {"service": "ec2"}, ["ec2", "cost optimization", "unused"]),
 76 |         ("resource_inventory", {"service": "ec2", "region": "us-west-2"}, ["ec2", "in the us-west-2 region", "inventory"]),
 77 |         ("resource_inventory", {"service": "s3"}, ["s3", "across all regions", "inventory"]),
 78 |         ("troubleshoot_service", {"service": "lambda", "resource_id": "my-function"}, ["lambda", "my-function", "troubleshoot"]),
 79 |         (
 80 |             "iam_policy_generator",
 81 |             {"service": "s3", "actions": "GetObject,PutObject", "resource_pattern": "arn:aws:s3:::my-bucket/*"},
 82 |             ["s3", "GetObject,PutObject", "arn:aws:s3:::my-bucket/*", "least-privilege"],
 83 |         ),
 84 |         ("service_monitoring", {"service": "rds", "metric_type": "performance"}, ["rds", "performance", "monitoring", "CloudWatch"]),
 85 |         ("disaster_recovery", {"service": "dynamodb", "recovery_point_objective": "15 minutes"}, ["dynamodb", "15 minutes", "disaster recovery"]),
 86 |         ("compliance_check", {"compliance_standard": "HIPAA", "service": "s3"}, ["HIPAA", "for s3", "compliance"]),
 87 |         ("resource_cleanup", {"service": "ec2", "criteria": "old"}, ["ec2", "old", "cleanup"]),
 88 |         # New prompts
 89 |         ("serverless_deployment", {"application_name": "test-app", "runtime": "python3.13"}, ["test-app", "python3.13", "serverless", "AWS SAM"]),
 90 |         ("container_orchestration", {"cluster_name": "test-cluster", "service_type": "fargate"}, ["test-cluster", "fargate", "container"]),
 91 |         ("vpc_network_design", {"vpc_name": "test-vpc", "cidr_block": "10.0.0.0/16"}, ["test-vpc", "10.0.0.0/16", "VPC", "security"]),
 92 |         ("infrastructure_automation", {"resource_type": "ec2", "automation_scope": "deployment"}, ["ec2", "deployment", "automation"]),
 93 |         ("security_posture_assessment", {}, ["Security Hub", "GuardDuty", "posture", "assessment"]),
 94 |         ("performance_tuning", {"service": "rds", "resource_id": "test-db"}, ["rds", "test-db", "performance", "metrics"]),
 95 |         ("multi_account_governance", {"account_type": "organization"}, ["organization", "multi-account", "governance"]),
 96 |     ],
 97 | )
 98 | def test_prompt_templates(prompt_functions, prompt_name, args, expected_content):
 99 |     """Test all prompt templates with various inputs using parametrized tests."""
100 |     # Get the captured function
101 |     prompt_func = prompt_functions.get(prompt_name)
102 |     assert prompt_func is not None, f"{prompt_name} prompt not found"
103 | 
104 |     # Test prompt output with the specified arguments
105 |     prompt_text = prompt_func(**args)
106 | 
107 |     # Check for expected content
108 |     for content in expected_content:
109 |         assert content.lower() in prompt_text.lower(), f"Expected '{content}' in {prompt_name} output"
110 | 
```

--------------------------------------------------------------------------------
/src/aws_mcp_server/server.py:
--------------------------------------------------------------------------------

```python
  1 | """Main server implementation for AWS MCP Server.
  2 | 
  3 | This module defines the MCP server instance and tool functions for AWS CLI interaction,
  4 | providing a standardized interface for AWS CLI command execution and documentation.
  5 | It also provides MCP Resources for AWS profiles, regions, and configuration.
  6 | """
  7 | 
  8 | import asyncio
  9 | import logging
 10 | import sys
 11 | 
 12 | from mcp.server.fastmcp import Context, FastMCP
 13 | from pydantic import Field
 14 | 
 15 | from aws_mcp_server import __version__
 16 | from aws_mcp_server.cli_executor import (
 17 |     CommandExecutionError,
 18 |     CommandHelpResult,
 19 |     CommandResult,
 20 |     CommandValidationError,
 21 |     check_aws_cli_installed,
 22 |     execute_aws_command,
 23 |     get_command_help,
 24 | )
 25 | from aws_mcp_server.config import INSTRUCTIONS
 26 | from aws_mcp_server.prompts import register_prompts
 27 | from aws_mcp_server.resources import register_resources
 28 | 
 29 | # Configure logging
 30 | logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler(sys.stderr)])
 31 | logger = logging.getLogger("aws-mcp-server")
 32 | 
 33 | 
 34 | # Run startup checks in synchronous context
 35 | def run_startup_checks():
 36 |     """Run startup checks to ensure AWS CLI is installed."""
 37 |     logger.info("Running startup checks...")
 38 |     if not asyncio.run(check_aws_cli_installed()):
 39 |         logger.error("AWS CLI is not installed or not in PATH. Please install AWS CLI.")
 40 |         sys.exit(1)
 41 |     logger.info("AWS CLI is installed and available")
 42 | 
 43 | 
 44 | # Call the checks
 45 | run_startup_checks()
 46 | 
 47 | # Create the FastMCP server following FastMCP best practices
 48 | mcp = FastMCP(
 49 |     "AWS MCP Server",
 50 |     instructions=INSTRUCTIONS,
 51 |     version=__version__,
 52 |     capabilities={"resources": {}},  # Enable resources capability
 53 | )
 54 | 
 55 | # Register prompt templates
 56 | register_prompts(mcp)
 57 | 
 58 | # Register AWS resources
 59 | register_resources(mcp)
 60 | 
 61 | 
 62 | @mcp.tool()
 63 | async def aws_cli_help(
 64 |     service: str = Field(description="AWS service (e.g., s3, ec2)"),
 65 |     command: str | None = Field(description="Command within the service", default=None),
 66 |     ctx: Context | None = None,
 67 | ) -> CommandHelpResult:
 68 |     """Get AWS CLI command documentation.
 69 | 
 70 |     Retrieves the help documentation for a specified AWS service or command
 71 |     by executing the 'aws <service> [command] help' command.
 72 | 
 73 |     Returns:
 74 |         CommandHelpResult containing the help text
 75 |     """
 76 |     logger.info(f"Getting documentation for service: {service}, command: {command or 'None'}")
 77 | 
 78 |     try:
 79 |         if ctx:
 80 |             await ctx.info(f"Fetching help for AWS {service} {command or ''}")
 81 | 
 82 |         # Reuse the get_command_help function from cli_executor
 83 |         result = await get_command_help(service, command)
 84 |         return result
 85 |     except Exception as e:
 86 |         logger.error(f"Error in aws_cli_help: {e}")
 87 |         return CommandHelpResult(help_text=f"Error retrieving help: {str(e)}")
 88 | 
 89 | 
 90 | @mcp.tool()
 91 | async def aws_cli_pipeline(
 92 |     command: str = Field(description="Complete AWS CLI command to execute (can include pipes with Unix commands)"),
 93 |     timeout: int | None = Field(description="Timeout in seconds (defaults to AWS_MCP_TIMEOUT)", default=None),
 94 |     ctx: Context | None = None,
 95 | ) -> CommandResult:
 96 |     """Execute an AWS CLI command, optionally with Unix command pipes.
 97 | 
 98 |     Validates, executes, and processes the results of an AWS CLI command,
 99 |     handling errors and formatting the output for better readability.
100 | 
101 |     The command can include Unix pipes (|) to filter or transform the output,
102 |     similar to a regular shell. The first command must be an AWS CLI command,
103 |     and subsequent piped commands must be basic Unix utilities.
104 | 
105 |     Supported Unix commands in pipes:
106 |     - File operations: ls, cat, cd, pwd, cp, mv, rm, mkdir, touch, chmod, chown
107 |     - Text processing: grep, sed, awk, cut, sort, uniq, wc, head, tail, tr, find
108 |     - System tools: ps, top, df, du, uname, whoami, date, which, echo
109 |     - Network tools: ping, ifconfig, netstat, curl, wget, dig, nslookup, ssh, scp
110 |     - Other utilities: man, less, tar, gzip, zip, xargs, jq, tee
111 | 
112 |     Examples:
113 |     - aws s3api list-buckets --query 'Buckets[*].Name' --output text
114 |     - aws s3api list-buckets --query 'Buckets[*].Name' --output text | sort
115 |     - aws ec2 describe-instances | grep InstanceId | wc -l
116 | 
117 |     Returns:
118 |         CommandResult containing output and status
119 |     """
120 |     logger.info(f"Executing command: {command}" + (f" with timeout: {timeout}" if timeout else ""))
121 | 
122 |     if ctx:
123 |         is_pipe = "|" in command
124 |         message = "Executing" + (" piped" if is_pipe else "") + " AWS CLI command"
125 |         await ctx.info(message + (f" with timeout: {timeout}s" if timeout else ""))
126 | 
127 |     try:
128 |         result = await execute_aws_command(command, timeout)
129 | 
130 |         # Format the output for better readability
131 |         if result["status"] == "success":
132 |             if ctx:
133 |                 await ctx.info("Command executed successfully")
134 |         else:
135 |             if ctx:
136 |                 await ctx.warning("Command failed")
137 | 
138 |         return CommandResult(status=result["status"], output=result["output"])
139 |     except CommandValidationError as e:
140 |         logger.warning(f"Command validation error: {e}")
141 |         return CommandResult(status="error", output=f"Command validation error: {str(e)}")
142 |     except CommandExecutionError as e:
143 |         logger.warning(f"Command execution error: {e}")
144 |         return CommandResult(status="error", output=f"Command execution error: {str(e)}")
145 |     except Exception as e:
146 |         logger.error(f"Error in aws_cli_pipeline: {e}")
147 |         return CommandResult(status="error", output=f"Unexpected error: {str(e)}")
148 | 
```

--------------------------------------------------------------------------------
/tests/integration/test_security_integration.py:
--------------------------------------------------------------------------------

```python
  1 | """Integration tests for security rules in AWS MCP Server.
  2 | 
  3 | These tests verify that security rules properly prevent dangerous commands
  4 | while allowing safe operations.
  5 | """
  6 | 
  7 | import pytest
  8 | 
  9 | from aws_mcp_server.server import aws_cli_pipeline
 10 | 
 11 | 
 12 | class TestSecurityIntegration:
 13 |     """Integration tests for security system.
 14 | 
 15 |     These tests validate that:
 16 |     1. Safe operations are allowed
 17 |     2. Dangerous operations are blocked
 18 |     3. Pipe commands are properly validated
 19 |     """
 20 | 
 21 |     @pytest.mark.asyncio
 22 |     @pytest.mark.integration
 23 |     @pytest.mark.parametrize(
 24 |         "command,should_succeed,expected_message",
 25 |         [
 26 |             # Safe operations that should succeed
 27 |             ("aws s3 ls", True, None),
 28 |             ("aws ec2 describe-instances", True, None),
 29 |             ("aws iam list-users", True, None),
 30 |             # Dangerous IAM operations that should be blocked
 31 |             (
 32 |                 "aws iam create-user --user-name test-user-12345",
 33 |                 False,
 34 |                 "restricted for security reasons",
 35 |             ),
 36 |             (
 37 |                 "aws iam create-access-key --user-name admin",
 38 |                 False,
 39 |                 "restricted for security reasons",
 40 |             ),
 41 |             # Dangerous CloudTrail operations (good for testing as they're security-related but not destructive)
 42 |             (
 43 |                 "aws cloudtrail delete-trail --name test-trail",
 44 |                 False,
 45 |                 "restricted for security reasons",
 46 |             ),
 47 |             # Complex regex pattern tests
 48 |             (
 49 |                 "aws iam create-user --user-name admin-user12345",
 50 |                 False,
 51 |                 "Creating users with sensitive names",
 52 |             ),
 53 |             (
 54 |                 "aws ec2 authorize-security-group-ingress --group-id sg-12345 --protocol tcp --port 22 --cidr 0.0.0.0/0",
 55 |                 False,
 56 |                 "restricted for security reasons",
 57 |             ),
 58 |             # Commands with safe overrides
 59 |             (
 60 |                 "aws iam create-user --help",
 61 |                 True,
 62 |                 None,
 63 |             ),
 64 |             (
 65 |                 "aws ec2 describe-security-groups",
 66 |                 True,
 67 |                 None,
 68 |             ),
 69 |         ],
 70 |     )
 71 |     async def test_security_rules(self, ensure_aws_credentials, command, should_succeed, expected_message):
 72 |         """Test that security rules block dangerous commands and allow safe operations.
 73 | 
 74 |         This test verifies each command against security rules without actually executing them
 75 |         against AWS services.
 76 |         """
 77 |         # Execute the command
 78 |         result = await aws_cli_pipeline(command=command, timeout=None, ctx=None)
 79 | 
 80 |         if should_succeed:
 81 |             if result["status"] != "success":
 82 |                 # If command would succeed but API returns error (e.g., invalid resource),
 83 |                 # we still want to verify it wasn't blocked by security rules
 84 |                 assert "restricted for security reasons" not in result["output"], f"Command should pass security validation but was blocked: {result['output']}"
 85 |                 assert "Command validation error" not in result["output"], f"Command should pass security validation but failed validation: {result['output']}"
 86 |             else:
 87 |                 assert result["status"] == "success", f"Command should succeed but failed: {result['output']}"
 88 |         else:
 89 |             assert result["status"] == "error", f"Command should fail but succeeded: {result['output']}"
 90 |             assert expected_message in result["output"], f"Expected error message '{expected_message}' not found in: {result['output']}"
 91 | 
 92 |     @pytest.mark.asyncio
 93 |     @pytest.mark.integration
 94 |     @pytest.mark.parametrize(
 95 |         "command,should_succeed,expected_message",
 96 |         [
 97 |             # Safe pipe commands
 98 |             (
 99 |                 "aws ec2 describe-regions --output text | grep us-east",
100 |                 True,
101 |                 None,
102 |             ),
103 |             (
104 |                 "aws s3 ls | grep bucket | wc -l",
105 |                 True,
106 |                 None,
107 |             ),
108 |             # Dangerous first command
109 |             (
110 |                 "aws iam create-user --user-name test-user-12345 | grep test",
111 |                 False,
112 |                 "restricted for security reasons",
113 |             ),
114 |             # Unsafe pipe command
115 |             (
116 |                 "aws s3 ls | sudo",  # sudo shouldn't be allowed in the allowed unix command list
117 |                 False,
118 |                 "not allowed",
119 |             ),
120 |             # Complex pipe chain
121 |             (
122 |                 "aws ec2 describe-regions --output json | grep RegionName | head -5 | sort",
123 |                 True,
124 |                 None,
125 |             ),
126 |         ],
127 |     )
128 |     async def test_piped_command_security(self, ensure_aws_credentials, command, should_succeed, expected_message):
129 |         """Test that security rules properly validate piped commands."""
130 |         result = await aws_cli_pipeline(command=command, timeout=None, ctx=None)
131 | 
132 |         if should_succeed:
133 |             if result["status"] != "success":
134 |                 # If command should be allowed but failed for other reasons,
135 |                 # verify it wasn't blocked by security rules
136 |                 assert "restricted for security reasons" not in result["output"], f"Command should pass security validation but was blocked: {result['output']}"
137 |                 assert "not allowed" not in result["output"], f"Command should pass security validation but was blocked: {result['output']}"
138 |         else:
139 |             assert result["status"] == "error", f"Command should fail but succeeded: {result['output']}"
140 |             assert expected_message in result["output"], f"Expected error message '{expected_message}' not found in: {result['output']}"
141 | 
```

--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------

```python
  1 | """Configuration for pytest."""
  2 | 
  3 | import os
  4 | 
  5 | import pytest
  6 | 
  7 | 
  8 | def pytest_addoption(parser):
  9 |     """Add command-line options to pytest."""
 10 |     parser.addoption(
 11 |         "--run-integration",
 12 |         action="store_true",
 13 |         default=False,
 14 |         help="Run integration tests that require AWS CLI and AWS account",
 15 |     )
 16 | 
 17 | 
 18 | def pytest_configure(config):
 19 |     """Register custom markers."""
 20 |     config.addinivalue_line("markers", "integration: mark test as requiring AWS CLI and AWS account")
 21 | 
 22 | 
 23 | def pytest_collection_modifyitems(config, items):
 24 |     """Skip integration tests unless --run-integration is specified."""
 25 |     print(f"Run integration flag: {config.getoption('--run-integration')}")
 26 | 
 27 |     if config.getoption("--run-integration"):
 28 |         # Run all tests
 29 |         print("Integration tests will be run")
 30 |         return
 31 | 
 32 |     skip_integration = pytest.mark.skip(reason="Integration tests need --run-integration option")
 33 |     print(f"Will check {len(items)} items for integration markers")
 34 | 
 35 |     for item in items:
 36 |         print(f"Test: {item.name}, keywords: {list(item.keywords)}")
 37 |         if "integration" in item.keywords:
 38 |             print(f"Skipping integration test: {item.name}")
 39 |             item.add_marker(skip_integration)
 40 | 
 41 | 
 42 | @pytest.fixture(scope="function")
 43 | async def aws_s3_bucket(ensure_aws_credentials):
 44 |     """Create or use an S3 bucket for integration tests.
 45 | 
 46 |     Uses AWS_TEST_BUCKET if specified, otherwise creates a temporary bucket
 47 |     and cleans it up after tests complete.
 48 |     """
 49 |     import asyncio
 50 |     import time
 51 |     import uuid
 52 | 
 53 |     from aws_mcp_server.server import aws_cli_pipeline
 54 | 
 55 |     print("AWS S3 bucket fixture called")
 56 | 
 57 |     # Use specified bucket or create a dynamically named one
 58 |     bucket_name = os.environ.get("AWS_TEST_BUCKET")
 59 |     bucket_created = False
 60 | 
 61 |     # Get region from environment or use configured default
 62 |     region = os.environ.get("AWS_TEST_REGION", os.environ.get("AWS_REGION", "us-east-1"))
 63 |     print(f"Using AWS region: {region}")
 64 | 
 65 |     print(f"Using bucket name: {bucket_name or 'Will create dynamic bucket'}")
 66 | 
 67 |     if not bucket_name:
 68 |         # Generate a unique bucket name with timestamp and random id
 69 |         timestamp = int(time.time())
 70 |         random_id = str(uuid.uuid4())[:8]
 71 |         bucket_name = f"aws-mcp-test-{timestamp}-{random_id}"
 72 |         print(f"Generated bucket name: {bucket_name}")
 73 | 
 74 |         # Create the bucket with region specified
 75 |         create_cmd = f"aws s3 mb s3://{bucket_name} --region {region}"
 76 |         print(f"Creating bucket with command: {create_cmd}")
 77 |         result = await aws_cli_pipeline(command=create_cmd, timeout=None, ctx=None)
 78 |         if result["status"] != "success":
 79 |             print(f"Failed to create bucket: {result['output']}")
 80 |             pytest.skip(f"Failed to create test bucket: {result['output']}")
 81 |         bucket_created = True
 82 |         print("Bucket created successfully")
 83 |         # Wait a moment for bucket to be fully available
 84 |         await asyncio.sleep(3)
 85 | 
 86 |     # Yield the bucket name for tests to use
 87 |     print(f"Yielding bucket name: {bucket_name}")
 88 |     yield bucket_name
 89 | 
 90 |     # Clean up the bucket if we created it
 91 |     if bucket_created:
 92 |         print(f"Cleaning up bucket: {bucket_name}")
 93 |         try:
 94 |             # First remove all objects
 95 |             print("Removing objects from bucket")
 96 |             await aws_cli_pipeline(command=f"aws s3 rm s3://{bucket_name} --recursive --region {region}", timeout=None, ctx=None)
 97 |             # Then delete the bucket
 98 |             print("Deleting bucket")
 99 |             await aws_cli_pipeline(command=f"aws s3 rb s3://{bucket_name} --region {region}", timeout=None, ctx=None)
100 |             print("Bucket cleanup complete")
101 |         except Exception as e:
102 |             print(f"Warning: Error cleaning up test bucket: {e}")
103 | 
104 | 
105 | @pytest.fixture
106 | def ensure_aws_credentials():
107 |     """Ensure AWS credentials are configured and AWS CLI is installed."""
108 |     import subprocess
109 | 
110 |     print("Checking AWS credentials and CLI")
111 | 
112 |     # Check for AWS CLI installation
113 |     try:
114 |         result = subprocess.run(["aws", "--version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
115 |         print(f"AWS CLI check: {result.returncode == 0}")
116 |         if result.returncode != 0:
117 |             print(f"AWS CLI not found: {result.stderr.decode('utf-8')}")
118 |             pytest.skip("AWS CLI not installed or not in PATH")
119 |     except (subprocess.SubprocessError, FileNotFoundError) as e:
120 |         print(f"AWS CLI check error: {str(e)}")
121 |         pytest.skip("AWS CLI not installed or not in PATH")
122 | 
123 |     # Check for AWS credentials - simplified check
124 |     home_dir = os.path.expanduser("~")
125 |     creds_file = os.path.join(home_dir, ".aws", "credentials")
126 |     config_file = os.path.join(home_dir, ".aws", "config")
127 | 
128 |     has_creds = os.path.exists(creds_file)
129 |     has_config = os.path.exists(config_file)
130 |     print(f"AWS files: credentials={has_creds}, config={has_config}")
131 |     # Don't skip based on file presence - let the get-caller-identity check decide
132 | 
133 |     # Verify AWS credentials work by making a simple call
134 |     try:
135 |         result = subprocess.run(["aws", "sts", "get-caller-identity"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=5, check=False)
136 |         print(f"AWS auth check: {result.returncode == 0}")
137 |         if result.returncode != 0:
138 |             error_msg = result.stderr.decode("utf-8")
139 |             print(f"AWS auth failed: {error_msg}")
140 |             pytest.skip(f"AWS credentials not valid: {error_msg}")
141 |         else:
142 |             print(f"AWS identity: {result.stdout.decode('utf-8')}")
143 |     except subprocess.SubprocessError as e:
144 |         print(f"AWS auth check error: {str(e)}")
145 |         pytest.skip("Failed to verify AWS credentials")
146 | 
147 |     # All checks passed - AWS CLI and credentials are working
148 |     print("AWS credentials verification successful")
149 |     return True
150 | 
```

--------------------------------------------------------------------------------
/tests/integration/test_server_integration.py:
--------------------------------------------------------------------------------

```python
  1 | """Mocked integration tests for AWS MCP Server functionality.
  2 | 
  3 | These tests use mocks rather than actual AWS CLI calls, so they can
  4 | run without AWS credentials or AWS CLI installed.
  5 | """
  6 | 
  7 | import json
  8 | import logging
  9 | import os
 10 | from unittest.mock import patch
 11 | 
 12 | import pytest
 13 | 
 14 | from aws_mcp_server.server import aws_cli_help, aws_cli_pipeline, mcp
 15 | 
 16 | # Enable debug logging for tests
 17 | logging.basicConfig(level=logging.DEBUG)
 18 | 
 19 | 
 20 | @pytest.fixture
 21 | def mock_aws_environment():
 22 |     """Set up mock AWS environment variables for testing."""
 23 |     original_env = os.environ.copy()
 24 |     os.environ["AWS_PROFILE"] = "test-profile"
 25 |     os.environ["AWS_REGION"] = "us-west-2"
 26 |     yield
 27 |     # Restore original environment
 28 |     os.environ.clear()
 29 |     os.environ.update(original_env)
 30 | 
 31 | 
 32 | @pytest.fixture
 33 | def mcp_client():
 34 |     """Return a FastMCP client for testing."""
 35 |     return mcp
 36 | 
 37 | 
 38 | class TestServerIntegration:
 39 |     """Integration tests for the AWS MCP Server using mocks.
 40 | 
 41 |     These tests use mocks and don't actually call AWS, but they test
 42 |     more of the system together than unit tests. They don't require the
 43 |     integration marker since they can run without AWS CLI or credentials."""
 44 | 
 45 |     @pytest.mark.asyncio
 46 |     @pytest.mark.parametrize(
 47 |         "service,command,mock_response,expected_content",
 48 |         [
 49 |             # Basic service help
 50 |             ("s3", None, {"help_text": "AWS S3 HELP\nCommands:\ncp\nls\nmv\nrm\nsync"}, ["AWS S3 HELP", "Commands", "ls", "sync"]),
 51 |             # Command-specific help
 52 |             (
 53 |                 "ec2",
 54 |                 "describe-instances",
 55 |                 {"help_text": "DESCRIPTION\n  Describes the specified instances.\n\nSYNOPSIS\n  describe-instances\n  [--instance-ids <value>]"},
 56 |                 ["DESCRIPTION", "SYNOPSIS", "instance-ids"],
 57 |             ),
 58 |             # Help for a different service
 59 |             ("lambda", "list-functions", {"help_text": "LAMBDA LIST-FUNCTIONS\nLists your Lambda functions"}, ["LAMBDA", "LIST-FUNCTIONS", "Lists"]),
 60 |         ],
 61 |     )
 62 |     @patch("aws_mcp_server.server.get_command_help")
 63 |     async def test_aws_cli_help_integration(self, mock_get_help, mock_aws_environment, service, command, mock_response, expected_content):
 64 |         """Test the aws_cli_help functionality with table-driven tests."""
 65 |         # Configure the mock response
 66 |         mock_get_help.return_value = mock_response
 67 | 
 68 |         # Call the aws_cli_help function
 69 |         result = await aws_cli_help(service=service, command=command, ctx=None)
 70 | 
 71 |         # Verify the results
 72 |         assert "help_text" in result
 73 |         for content in expected_content:
 74 |             assert content in result["help_text"], f"Expected '{content}' in help text"
 75 | 
 76 |         # Verify the mock was called correctly
 77 |         mock_get_help.assert_called_once_with(service, command)
 78 | 
 79 |     @pytest.mark.asyncio
 80 |     @pytest.mark.parametrize(
 81 |         "command,mock_response,expected_result,timeout",
 82 |         [
 83 |             # JSON output test
 84 |             (
 85 |                 "aws s3 ls --output json",
 86 |                 {"status": "success", "output": json.dumps({"Buckets": [{"Name": "test-bucket", "CreationDate": "2023-01-01T00:00:00Z"}]})},
 87 |                 {"status": "success", "contains": ["Buckets", "test-bucket"]},
 88 |                 None,
 89 |             ),
 90 |             # Text output test
 91 |             (
 92 |                 "aws ec2 describe-instances --query 'Reservations[*]' --output text",
 93 |                 {"status": "success", "output": "i-12345\trunning\tt2.micro"},
 94 |                 {"status": "success", "contains": ["i-12345", "running"]},
 95 |                 None,
 96 |             ),
 97 |             # Test with custom timeout
 98 |             ("aws rds describe-db-instances", {"status": "success", "output": "DB instances list"}, {"status": "success", "contains": ["DB instances"]}, 60),
 99 |             # Error case
100 |             (
101 |                 "aws s3 ls --invalid-flag",
102 |                 {"status": "error", "output": "Unknown options: --invalid-flag"},
103 |                 {"status": "error", "contains": ["--invalid-flag"]},
104 |                 None,
105 |             ),
106 |             # Piped command
107 |             (
108 |                 "aws s3api list-buckets --query 'Buckets[*].Name' --output text | sort",
109 |                 {"status": "success", "output": "bucket1\nbucket2\nbucket3"},
110 |                 {"status": "success", "contains": ["bucket1", "bucket3"]},
111 |                 None,
112 |             ),
113 |         ],
114 |     )
115 |     @patch("aws_mcp_server.server.execute_aws_command")
116 |     async def test_aws_cli_pipeline_scenarios(self, mock_execute, mock_aws_environment, command, mock_response, expected_result, timeout):
117 |         """Test aws_cli_pipeline with various scenarios using table-driven tests."""
118 |         # Configure the mock response
119 |         mock_execute.return_value = mock_response
120 | 
121 |         # Call the aws_cli_pipeline function
122 |         result = await aws_cli_pipeline(command=command, timeout=timeout, ctx=None)
123 | 
124 |         # Verify status
125 |         assert result["status"] == expected_result["status"]
126 | 
127 |         # Verify expected content is present
128 |         for content in expected_result["contains"]:
129 |             assert content in result["output"], f"Expected '{content}' in output"
130 | 
131 |         # Verify the mock was called correctly
132 |         mock_execute.assert_called_once_with(command, timeout)
133 | 
134 |     @pytest.mark.asyncio
135 |     @patch("aws_mcp_server.resources.get_aws_profiles")
136 |     @patch("aws_mcp_server.resources.get_aws_regions")
137 |     @patch("aws_mcp_server.resources.get_aws_environment")
138 |     @patch("aws_mcp_server.resources.get_aws_account_info")
139 |     async def test_mcp_resources_access(
140 |         self, mock_get_aws_account_info, mock_get_aws_environment, mock_get_aws_regions, mock_get_aws_profiles, mock_aws_environment, mcp_client
141 |     ):
142 |         """Test that MCP resources are properly registered and accessible to clients."""
143 |         # Set up mock return values
144 |         mock_get_aws_profiles.return_value = ["default", "test-profile", "dev"]
145 |         mock_get_aws_regions.return_value = [
146 |             {"RegionName": "us-east-1", "RegionDescription": "US East (N. Virginia)"},
147 |             {"RegionName": "us-west-2", "RegionDescription": "US West (Oregon)"},
148 |         ]
149 |         mock_get_aws_environment.return_value = {
150 |             "aws_profile": "test-profile",
151 |             "aws_region": "us-west-2",
152 |             "has_credentials": True,
153 |             "credentials_source": "profile",
154 |         }
155 |         mock_get_aws_account_info.return_value = {
156 |             "account_id": "123456789012",
157 |             "account_alias": "test-account",
158 |             "organization_id": "o-abcdef123456",
159 |         }
160 | 
161 |         # Define the expected resource URIs
162 |         expected_resources = ["aws://config/profiles", "aws://config/regions", "aws://config/environment", "aws://config/account"]
163 | 
164 |         # Test that resources are accessible through MCP client
165 |         resources = await mcp_client.list_resources()
166 | 
167 |         # Verify all expected resources are present
168 |         resource_uris = [str(r.uri) for r in resources]
169 |         for uri in expected_resources:
170 |             assert uri in resource_uris, f"Resource {uri} not found in resources list"
171 | 
172 |         # Test accessing each resource by URI
173 |         for uri in expected_resources:
174 |             resource = await mcp_client.read_resource(uri=uri)
175 |             assert resource is not None, f"Failed to read resource {uri}"
176 | 
177 |             # Resource is a list with one item that has a content attribute
178 |             # The content is a JSON string that needs to be parsed
179 |             import json
180 | 
181 |             content = json.loads(resource[0].content)
182 | 
183 |             # Verify specific resource content
184 |             if uri == "aws://config/profiles":
185 |                 assert "profiles" in content
186 |                 assert len(content["profiles"]) == 3
187 |                 assert any(p["name"] == "test-profile" and p["is_current"] for p in content["profiles"])
188 | 
189 |             elif uri == "aws://config/regions":
190 |                 assert "regions" in content
191 |                 assert len(content["regions"]) == 2
192 |                 assert any(r["name"] == "us-west-2" and r["is_current"] for r in content["regions"])
193 | 
194 |             elif uri == "aws://config/environment":
195 |                 assert content["aws_profile"] == "test-profile"
196 |                 assert content["aws_region"] == "us-west-2"
197 |                 assert content["has_credentials"] is True
198 | 
199 |             elif uri == "aws://config/account":
200 |                 assert content["account_id"] == "123456789012"
201 |                 assert content["account_alias"] == "test-account"
202 | 
```

--------------------------------------------------------------------------------
/src/aws_mcp_server/tools.py:
--------------------------------------------------------------------------------

```python
  1 | """Command execution utilities for AWS MCP Server.
  2 | 
  3 | This module provides utilities for validating and executing commands, including:
  4 | - AWS CLI commands
  5 | - Basic Unix commands
  6 | - Command pipes (piping output from one command to another)
  7 | """
  8 | 
  9 | import asyncio
 10 | import logging
 11 | import shlex
 12 | from typing import List, TypedDict
 13 | 
 14 | from aws_mcp_server.config import DEFAULT_TIMEOUT, MAX_OUTPUT_SIZE
 15 | 
 16 | # Configure module logger
 17 | logger = logging.getLogger(__name__)
 18 | 
 19 | # List of allowed Unix commands that can be used in a pipe
 20 | ALLOWED_UNIX_COMMANDS = [
 21 |     # File operations
 22 |     "cat",
 23 |     "ls",
 24 |     "cd",
 25 |     "pwd",
 26 |     "cp",
 27 |     "mv",
 28 |     "rm",
 29 |     "mkdir",
 30 |     "touch",
 31 |     "chmod",
 32 |     "chown",
 33 |     # Text processing
 34 |     "grep",
 35 |     "sed",
 36 |     "awk",
 37 |     "cut",
 38 |     "sort",
 39 |     "uniq",
 40 |     "wc",
 41 |     "head",
 42 |     "tail",
 43 |     "tr",
 44 |     "find",
 45 |     # System information
 46 |     "ps",
 47 |     "top",
 48 |     "df",
 49 |     "du",
 50 |     "uname",
 51 |     "whoami",
 52 |     "date",
 53 |     "which",
 54 |     "echo",
 55 |     # Networking
 56 |     "ping",
 57 |     "ifconfig",
 58 |     "netstat",
 59 |     "curl",
 60 |     "wget",
 61 |     "dig",
 62 |     "nslookup",
 63 |     "ssh",
 64 |     "scp",
 65 |     # Other utilities
 66 |     "man",
 67 |     "less",
 68 |     "tar",
 69 |     "gzip",
 70 |     "gunzip",
 71 |     "zip",
 72 |     "unzip",
 73 |     "xargs",
 74 |     "jq",
 75 |     "tee",
 76 | ]
 77 | 
 78 | 
 79 | class CommandResult(TypedDict):
 80 |     """Type definition for command execution results."""
 81 | 
 82 |     status: str
 83 |     output: str
 84 | 
 85 | 
 86 | def validate_unix_command(command: str) -> bool:
 87 |     """Validate that a command is an allowed Unix command.
 88 | 
 89 |     Args:
 90 |         command: The Unix command to validate
 91 | 
 92 |     Returns:
 93 |         True if the command is valid, False otherwise
 94 |     """
 95 |     cmd_parts = shlex.split(command)
 96 |     if not cmd_parts:
 97 |         return False
 98 | 
 99 |     # Check if the command is in the allowed list
100 |     return cmd_parts[0] in ALLOWED_UNIX_COMMANDS
101 | 
102 | 
103 | def is_pipe_command(command: str) -> bool:
104 |     """Check if a command contains a pipe operator.
105 | 
106 |     Args:
107 |         command: The command to check
108 | 
109 |     Returns:
110 |         True if the command contains a pipe operator, False otherwise
111 |     """
112 |     # Check for pipe operator that's not inside quotes
113 |     in_single_quote = False
114 |     in_double_quote = False
115 |     escaped = False
116 | 
117 |     for _, char in enumerate(command):
118 |         # Handle escape sequences
119 |         if char == "\\" and not escaped:
120 |             escaped = True
121 |             continue
122 | 
123 |         if not escaped:
124 |             if char == "'" and not in_double_quote:
125 |                 in_single_quote = not in_single_quote
126 |             elif char == '"' and not in_single_quote:
127 |                 in_double_quote = not in_double_quote
128 |             elif char == "|" and not in_single_quote and not in_double_quote:
129 |                 return True
130 | 
131 |         escaped = False
132 | 
133 |     return False
134 | 
135 | 
136 | def split_pipe_command(pipe_command: str) -> List[str]:
137 |     """Split a piped command into individual commands.
138 | 
139 |     Args:
140 |         pipe_command: The piped command string
141 | 
142 |     Returns:
143 |         List of individual command strings
144 |     """
145 |     commands = []
146 |     current_command = ""
147 |     in_single_quote = False
148 |     in_double_quote = False
149 |     escaped = False
150 | 
151 |     for _, char in enumerate(pipe_command):
152 |         # Handle escape sequences
153 |         if char == "\\" and not escaped:
154 |             escaped = True
155 |             current_command += char
156 |             continue
157 | 
158 |         if not escaped:
159 |             if char == "'" and not in_double_quote:
160 |                 in_single_quote = not in_single_quote
161 |                 current_command += char
162 |             elif char == '"' and not in_single_quote:
163 |                 in_double_quote = not in_double_quote
164 |                 current_command += char
165 |             elif char == "|" and not in_single_quote and not in_double_quote:
166 |                 commands.append(current_command.strip())
167 |                 current_command = ""
168 |             else:
169 |                 current_command += char
170 |         else:
171 |             # Add the escaped character
172 |             current_command += char
173 |             escaped = False
174 | 
175 |     if current_command.strip():
176 |         commands.append(current_command.strip())
177 | 
178 |     return commands
179 | 
180 | 
181 | async def execute_piped_command(pipe_command: str, timeout: int | None = None) -> CommandResult:
182 |     """Execute a command that contains pipes.
183 | 
184 |     Args:
185 |         pipe_command: The piped command to execute
186 |         timeout: Optional timeout in seconds (defaults to DEFAULT_TIMEOUT)
187 | 
188 |     Returns:
189 |         CommandResult containing output and status
190 |     """
191 |     # Set timeout
192 |     if timeout is None:
193 |         timeout = DEFAULT_TIMEOUT
194 | 
195 |     logger.debug(f"Executing piped command: {pipe_command}")
196 | 
197 |     try:
198 |         # Split the pipe_command into individual commands
199 |         commands = split_pipe_command(pipe_command)
200 | 
201 |         # For each command, split it into command parts for subprocess_exec
202 |         command_parts_list = [shlex.split(cmd) for cmd in commands]
203 | 
204 |         if len(commands) == 0:
205 |             return CommandResult(status="error", output="Empty command")
206 | 
207 |         # Execute the first command
208 |         first_cmd = command_parts_list[0]
209 |         first_process = await asyncio.create_subprocess_exec(*first_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
210 | 
211 |         current_process = first_process
212 |         current_stdout = None
213 |         current_stderr = None
214 | 
215 |         # For each additional command in the pipe, execute it with the previous command's output
216 |         for cmd_parts in command_parts_list[1:]:
217 |             try:
218 |                 # Wait for the previous command to complete with timeout
219 |                 current_stdout, current_stderr = await asyncio.wait_for(current_process.communicate(), timeout)
220 | 
221 |                 if current_process.returncode != 0:
222 |                     # If previous command failed, stop the pipe execution
223 |                     stderr_str = current_stderr.decode("utf-8", errors="replace")
224 |                     logger.warning(f"Piped command failed with return code {current_process.returncode}: {pipe_command}")
225 |                     logger.debug(f"Command error output: {stderr_str}")
226 |                     return CommandResult(status="error", output=stderr_str or "Command failed with no error output")
227 | 
228 |                 # Create the next process with the previous output as input
229 |                 next_process = await asyncio.create_subprocess_exec(
230 |                     *cmd_parts, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
231 |                 )
232 | 
233 |                 # Pass the output of the previous command to the input of the next command
234 |                 stdout, stderr = await asyncio.wait_for(next_process.communicate(input=current_stdout), timeout)
235 | 
236 |                 current_process = next_process
237 |                 current_stdout = stdout
238 |                 current_stderr = stderr
239 | 
240 |             except asyncio.TimeoutError:
241 |                 logger.warning(f"Piped command timed out after {timeout} seconds: {pipe_command}")
242 |                 try:
243 |                     # process.kill() is synchronous, not a coroutine
244 |                     current_process.kill()
245 |                 except Exception as e:
246 |                     logger.error(f"Error killing process: {e}")
247 |                 return CommandResult(status="error", output=f"Command timed out after {timeout} seconds")
248 | 
249 |         # Wait for the final command to complete if it hasn't already
250 |         if current_stdout is None:
251 |             try:
252 |                 current_stdout, current_stderr = await asyncio.wait_for(current_process.communicate(), timeout)
253 |             except asyncio.TimeoutError:
254 |                 logger.warning(f"Piped command timed out after {timeout} seconds: {pipe_command}")
255 |                 try:
256 |                     current_process.kill()
257 |                 except Exception as e:
258 |                     logger.error(f"Error killing process: {e}")
259 |                 return CommandResult(status="error", output=f"Command timed out after {timeout} seconds")
260 | 
261 |         # Process output
262 |         stdout_str = current_stdout.decode("utf-8", errors="replace")
263 |         stderr_str = current_stderr.decode("utf-8", errors="replace")
264 | 
265 |         # Truncate output if necessary
266 |         if len(stdout_str) > MAX_OUTPUT_SIZE:
267 |             logger.info(f"Output truncated from {len(stdout_str)} to {MAX_OUTPUT_SIZE} characters")
268 |             stdout_str = stdout_str[:MAX_OUTPUT_SIZE] + "\n... (output truncated)"
269 | 
270 |         if current_process.returncode != 0:
271 |             logger.warning(f"Piped command failed with return code {current_process.returncode}: {pipe_command}")
272 |             logger.debug(f"Command error output: {stderr_str}")
273 |             return CommandResult(status="error", output=stderr_str or "Command failed with no error output")
274 | 
275 |         return CommandResult(status="success", output=stdout_str)
276 |     except Exception as e:
277 |         logger.error(f"Failed to execute piped command: {str(e)}")
278 |         return CommandResult(status="error", output=f"Failed to execute command: {str(e)}")
279 | 
```

--------------------------------------------------------------------------------
/tests/unit/test_server.py:
--------------------------------------------------------------------------------

```python
  1 | """Tests for the FastMCP server implementation."""
  2 | 
  3 | from unittest.mock import ANY, AsyncMock, patch
  4 | 
  5 | import pytest
  6 | 
  7 | from aws_mcp_server.cli_executor import CommandExecutionError, CommandValidationError
  8 | from aws_mcp_server.server import aws_cli_help, aws_cli_pipeline, mcp, run_startup_checks
  9 | 
 10 | 
 11 | def test_run_startup_checks():
 12 |     """Test the run_startup_checks function."""
 13 |     # Create a complete mock for asyncio.run to avoid the coroutine warning
 14 |     # We'll mock both the check_aws_cli_installed and asyncio.run
 15 |     # This way we don't rely on any actual coroutine behavior in testing
 16 | 
 17 |     # Test when AWS CLI is installed
 18 |     with patch("aws_mcp_server.server.check_aws_cli_installed") as mock_check:
 19 |         # Don't use the actual coroutine
 20 |         mock_check.return_value = None  # Not used when mocking asyncio.run
 21 | 
 22 |         with patch("aws_mcp_server.server.asyncio.run", return_value=True):
 23 |             with patch("sys.exit") as mock_exit:
 24 |                 run_startup_checks()
 25 |                 mock_exit.assert_not_called()
 26 | 
 27 |     # Test when AWS CLI is not installed
 28 |     with patch("aws_mcp_server.server.check_aws_cli_installed") as mock_check:
 29 |         # Don't use the actual coroutine
 30 |         mock_check.return_value = None  # Not used when mocking asyncio.run
 31 | 
 32 |         with patch("aws_mcp_server.server.asyncio.run", return_value=False):
 33 |             with patch("sys.exit") as mock_exit:
 34 |                 run_startup_checks()
 35 |                 mock_exit.assert_called_once_with(1)
 36 | 
 37 | 
 38 | @pytest.mark.asyncio
 39 | @pytest.mark.parametrize(
 40 |     "service,command,expected_result",
 41 |     [
 42 |         ("s3", None, {"help_text": "Test help text"}),
 43 |         ("s3", "ls", {"help_text": "Test help text"}),
 44 |         ("ec2", "describe-instances", {"help_text": "Test help text"}),
 45 |     ],
 46 | )
 47 | async def test_aws_cli_help(service, command, expected_result):
 48 |     """Test the aws_cli_help tool with various inputs."""
 49 |     # Mock the get_command_help function instead of execute_aws_command
 50 |     with patch("aws_mcp_server.server.get_command_help", new_callable=AsyncMock) as mock_get_help:
 51 |         mock_get_help.return_value = expected_result
 52 | 
 53 |         # Call the tool with specified service and command
 54 |         result = await aws_cli_help(service=service, command=command)
 55 | 
 56 |         # Verify the result
 57 |         assert result == expected_result
 58 | 
 59 |         # Verify the correct arguments were passed to the mocked function
 60 |         mock_get_help.assert_called_with(service, command)
 61 | 
 62 | 
 63 | @pytest.mark.asyncio
 64 | async def test_aws_cli_help_with_context():
 65 |     """Test the aws_cli_help tool with context."""
 66 |     mock_ctx = AsyncMock()
 67 | 
 68 |     with patch("aws_mcp_server.server.get_command_help", new_callable=AsyncMock) as mock_get_help:
 69 |         mock_get_help.return_value = {"help_text": "Test help text"}
 70 | 
 71 |         result = await aws_cli_help(service="s3", command="ls", ctx=mock_ctx)
 72 | 
 73 |         assert result == {"help_text": "Test help text"}
 74 |         mock_ctx.info.assert_called_once()
 75 |         assert "Fetching help for AWS s3 ls" in mock_ctx.info.call_args[0][0]
 76 | 
 77 | 
 78 | @pytest.mark.asyncio
 79 | async def test_aws_cli_help_exception_handling():
 80 |     """Test exception handling in aws_cli_help."""
 81 |     with patch("aws_mcp_server.server.get_command_help", side_effect=Exception("Test exception")):
 82 |         result = await aws_cli_help(service="s3")
 83 | 
 84 |         assert "help_text" in result
 85 |         assert "Error retrieving help" in result["help_text"]
 86 |         assert "Test exception" in result["help_text"]
 87 | 
 88 | 
 89 | @pytest.mark.asyncio
 90 | @pytest.mark.parametrize(
 91 |     "command,timeout,expected_result",
 92 |     [
 93 |         # Basic success case
 94 |         ("aws s3 ls", None, {"status": "success", "output": "Test output"}),
 95 |         # Success with custom timeout
 96 |         ("aws s3 ls", 60, {"status": "success", "output": "Test output"}),
 97 |         # Complex command success
 98 |         ("aws ec2 describe-instances --filters Name=instance-state-name,Values=running", None, {"status": "success", "output": "Running instances"}),
 99 |     ],
100 | )
101 | async def test_aws_cli_pipeline_success(command, timeout, expected_result):
102 |     """Test the aws_cli_pipeline tool with successful execution."""
103 |     # Need to patch check_aws_cli_installed to avoid the coroutine warning
104 |     with patch("aws_mcp_server.server.check_aws_cli_installed", return_value=None):
105 |         # Mock the execute_aws_command function
106 |         with patch("aws_mcp_server.server.execute_aws_command", new_callable=AsyncMock) as mock_execute:
107 |             mock_execute.return_value = expected_result
108 | 
109 |             # Call the aws_cli_pipeline function
110 |             result = await aws_cli_pipeline(command=command, timeout=timeout)
111 | 
112 |             # Verify the result
113 |             assert result["status"] == expected_result["status"]
114 |             assert result["output"] == expected_result["output"]
115 | 
116 |             # Verify the correct arguments were passed to the mocked function
117 |             mock_execute.assert_called_with(command, timeout if timeout else ANY)
118 | 
119 | 
120 | @pytest.mark.asyncio
121 | async def test_aws_cli_pipeline_with_context():
122 |     """Test the aws_cli_pipeline tool with context."""
123 |     mock_ctx = AsyncMock()
124 | 
125 |     # Need to patch check_aws_cli_installed to avoid the coroutine warning
126 |     with patch("aws_mcp_server.server.check_aws_cli_installed", return_value=None):
127 |         # Test successful command with context
128 |         with patch("aws_mcp_server.server.execute_aws_command", new_callable=AsyncMock) as mock_execute:
129 |             mock_execute.return_value = {"status": "success", "output": "Test output"}
130 | 
131 |             result = await aws_cli_pipeline(command="aws s3 ls", ctx=mock_ctx)
132 | 
133 |             assert result["status"] == "success"
134 |             assert result["output"] == "Test output"
135 | 
136 |             # Verify context was used correctly
137 |             assert mock_ctx.info.call_count == 2
138 |             assert "Executing AWS CLI command" in mock_ctx.info.call_args_list[0][0][0]
139 |             assert "Command executed successfully" in mock_ctx.info.call_args_list[1][0][0]
140 | 
141 |         # Test failed command with context
142 |         mock_ctx.reset_mock()
143 |         with patch("aws_mcp_server.server.execute_aws_command", new_callable=AsyncMock) as mock_execute:
144 |             mock_execute.return_value = {"status": "error", "output": "Error output"}
145 | 
146 |             result = await aws_cli_pipeline(command="aws s3 ls", ctx=mock_ctx)
147 | 
148 |             assert result["status"] == "error"
149 |             assert result["output"] == "Error output"
150 | 
151 |             # Verify context was used correctly
152 |             assert mock_ctx.info.call_count == 1
153 |             assert mock_ctx.warning.call_count == 1
154 |             assert "Command failed" in mock_ctx.warning.call_args[0][0]
155 | 
156 | 
157 | @pytest.mark.asyncio
158 | async def test_aws_cli_pipeline_with_context_and_timeout():
159 |     """Test the aws_cli_pipeline tool with context and timeout."""
160 |     mock_ctx = AsyncMock()
161 | 
162 |     # Need to patch check_aws_cli_installed to avoid the coroutine warning
163 |     with patch("aws_mcp_server.server.check_aws_cli_installed", return_value=None):
164 |         with patch("aws_mcp_server.server.execute_aws_command", new_callable=AsyncMock) as mock_execute:
165 |             mock_execute.return_value = {"status": "success", "output": "Test output"}
166 | 
167 |             await aws_cli_pipeline(command="aws s3 ls", timeout=60, ctx=mock_ctx)
168 | 
169 |             # Verify timeout was mentioned in the context message
170 |             message = mock_ctx.info.call_args_list[0][0][0]
171 |             assert "with timeout: 60s" in message
172 | 
173 | 
174 | @pytest.mark.asyncio
175 | @pytest.mark.parametrize(
176 |     "command,exception,expected_error_type,expected_message",
177 |     [
178 |         # Validation error
179 |         ("not aws", CommandValidationError("Invalid command"), "Command validation error", "Invalid command"),
180 |         # Execution error
181 |         ("aws s3 ls", CommandExecutionError("Execution failed"), "Command execution error", "Execution failed"),
182 |         # Timeout error
183 |         ("aws ec2 describe-instances", CommandExecutionError("Command timed out"), "Command execution error", "Command timed out"),
184 |         # Generic/unexpected error
185 |         ("aws dynamodb scan", Exception("Unexpected error"), "Unexpected error", "Unexpected error"),
186 |     ],
187 | )
188 | async def test_aws_cli_pipeline_errors(command, exception, expected_error_type, expected_message):
189 |     """Test the aws_cli_pipeline tool with various error scenarios."""
190 |     # Need to patch check_aws_cli_installed to avoid the coroutine warning
191 |     with patch("aws_mcp_server.server.check_aws_cli_installed", return_value=None):
192 |         # Mock the execute_aws_command function to raise the specified exception
193 |         with patch("aws_mcp_server.server.execute_aws_command", side_effect=exception) as mock_execute:
194 |             # Call the tool
195 |             result = await aws_cli_pipeline(command=command)
196 | 
197 |             # Verify error status and message
198 |             assert result["status"] == "error"
199 |             assert expected_error_type in result["output"]
200 |             assert expected_message in result["output"]
201 | 
202 |             # Verify the command was called correctly
203 |             mock_execute.assert_called_with(command, ANY)
204 | 
205 | 
206 | @pytest.mark.asyncio
207 | async def test_mcp_server_initialization():
208 |     """Test that the MCP server initializes correctly."""
209 |     # Verify server was created with correct name
210 |     assert mcp.name == "AWS MCP Server"
211 | 
212 |     # Verify tools are registered by calling them
213 |     # This ensures the tools exist without depending on FastMCP's internal structure
214 |     assert callable(aws_cli_help)
215 |     assert callable(aws_cli_pipeline)
216 | 
```

--------------------------------------------------------------------------------
/src/aws_mcp_server/cli_executor.py:
--------------------------------------------------------------------------------

```python
  1 | """Utility for executing AWS CLI commands.
  2 | 
  3 | This module provides functions to validate and execute AWS CLI commands
  4 | with proper error handling, timeouts, and output processing.
  5 | """
  6 | 
  7 | import asyncio
  8 | import logging
  9 | import shlex
 10 | from typing import TypedDict
 11 | 
 12 | from aws_mcp_server.config import DEFAULT_TIMEOUT, MAX_OUTPUT_SIZE
 13 | from aws_mcp_server.security import validate_aws_command, validate_pipe_command
 14 | from aws_mcp_server.tools import (
 15 |     CommandResult,
 16 |     execute_piped_command,
 17 |     is_pipe_command,
 18 |     split_pipe_command,
 19 | )
 20 | 
 21 | # Configure module logger
 22 | logger = logging.getLogger(__name__)
 23 | 
 24 | 
 25 | class CommandHelpResult(TypedDict):
 26 |     """Type definition for command help results."""
 27 | 
 28 |     help_text: str
 29 | 
 30 | 
 31 | class CommandValidationError(Exception):
 32 |     """Exception raised when a command fails validation.
 33 | 
 34 |     This exception is raised when a command doesn't meet the
 35 |     validation requirements, such as starting with 'aws'.
 36 |     """
 37 | 
 38 |     pass
 39 | 
 40 | 
 41 | class CommandExecutionError(Exception):
 42 |     """Exception raised when a command fails to execute.
 43 | 
 44 |     This exception is raised when there's an error during command
 45 |     execution, such as timeouts or subprocess failures.
 46 |     """
 47 | 
 48 |     pass
 49 | 
 50 | 
 51 | def is_auth_error(error_output: str) -> bool:
 52 |     """Detect if an error is related to authentication.
 53 | 
 54 |     Args:
 55 |         error_output: The error output from AWS CLI
 56 | 
 57 |     Returns:
 58 |         True if the error is related to authentication, False otherwise
 59 |     """
 60 |     auth_error_patterns = [
 61 |         "Unable to locate credentials",
 62 |         "ExpiredToken",
 63 |         "AccessDenied",
 64 |         "AuthFailure",
 65 |         "The security token included in the request is invalid",
 66 |         "The config profile could not be found",
 67 |         "UnrecognizedClientException",
 68 |         "InvalidClientTokenId",
 69 |         "InvalidAccessKeyId",
 70 |         "SignatureDoesNotMatch",
 71 |         "Your credential profile is not properly configured",
 72 |         "credentials could not be refreshed",
 73 |         "NoCredentialProviders",
 74 |     ]
 75 |     return any(pattern in error_output for pattern in auth_error_patterns)
 76 | 
 77 | 
 78 | async def check_aws_cli_installed() -> bool:
 79 |     """Check if AWS CLI is installed and accessible.
 80 | 
 81 |     Returns:
 82 |         True if AWS CLI is installed, False otherwise
 83 |     """
 84 |     try:
 85 |         # Split command safely for exec
 86 |         cmd_parts = ["aws", "--version"]
 87 | 
 88 |         # Create subprocess using exec (safer than shell=True)
 89 |         process = await asyncio.create_subprocess_exec(*cmd_parts, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
 90 |         await process.communicate()
 91 |         return process.returncode == 0
 92 |     except Exception:
 93 |         return False
 94 | 
 95 | 
 96 | # Command validation functions are now imported from aws_mcp_server.security
 97 | 
 98 | 
 99 | async def execute_aws_command(command: str, timeout: int | None = None) -> CommandResult:
100 |     """Execute an AWS CLI command and return the result.
101 | 
102 |     Validates, executes, and processes the results of an AWS CLI command,
103 |     handling timeouts and output size limits.
104 | 
105 |     Args:
106 |         command: The AWS CLI command to execute (must start with 'aws')
107 |         timeout: Optional timeout in seconds (defaults to DEFAULT_TIMEOUT)
108 | 
109 |     Returns:
110 |         CommandResult containing output and status
111 | 
112 |     Raises:
113 |         CommandValidationError: If the command is invalid
114 |         CommandExecutionError: If the command fails to execute
115 |     """
116 |     # Check if this is a piped command
117 |     if is_pipe_command(command):
118 |         return await execute_pipe_command(command, timeout)
119 | 
120 |     # Validate the command
121 |     try:
122 |         validate_aws_command(command)
123 |     except ValueError as e:
124 |         raise CommandValidationError(str(e)) from e
125 | 
126 |     # Set timeout
127 |     if timeout is None:
128 |         timeout = DEFAULT_TIMEOUT
129 | 
130 |     # Check if the command needs a region and doesn't have one specified
131 |     from aws_mcp_server.config import AWS_REGION
132 | 
133 |     # Split by spaces and check for EC2 service specifically
134 |     cmd_parts = shlex.split(command)
135 |     is_ec2_command = len(cmd_parts) >= 2 and cmd_parts[0] == "aws" and cmd_parts[1] == "ec2"
136 |     has_region = "--region" in cmd_parts
137 | 
138 |     # If it's an EC2 command and doesn't have --region
139 |     if is_ec2_command and not has_region:
140 |         # Add the region parameter
141 |         command = f"{command} --region {AWS_REGION}"
142 |         logger.debug(f"Added region to command: {command}")
143 | 
144 |     logger.debug(f"Executing AWS command: {command}")
145 | 
146 |     try:
147 |         # Split command safely for exec
148 |         cmd_parts = shlex.split(command)
149 | 
150 |         # Create subprocess using exec (safer than shell=True)
151 |         process = await asyncio.create_subprocess_exec(*cmd_parts, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
152 | 
153 |         # Wait for the process to complete with timeout
154 |         try:
155 |             stdout, stderr = await asyncio.wait_for(process.communicate(), timeout)
156 |             logger.debug(f"Command completed with return code: {process.returncode}")
157 |         except asyncio.TimeoutError as timeout_error:
158 |             logger.warning(f"Command timed out after {timeout} seconds: {command}")
159 |             try:
160 |                 # process.kill() is synchronous, not a coroutine
161 |                 process.kill()
162 |             except Exception as e:
163 |                 logger.error(f"Error killing process: {e}")
164 |             raise CommandExecutionError(f"Command timed out after {timeout} seconds") from timeout_error
165 | 
166 |         # Process output
167 |         stdout_str = stdout.decode("utf-8", errors="replace")
168 |         stderr_str = stderr.decode("utf-8", errors="replace")
169 | 
170 |         # Truncate output if necessary
171 |         if len(stdout_str) > MAX_OUTPUT_SIZE:
172 |             logger.info(f"Output truncated from {len(stdout_str)} to {MAX_OUTPUT_SIZE} characters")
173 |             stdout_str = stdout_str[:MAX_OUTPUT_SIZE] + "\n... (output truncated)"
174 | 
175 |         if process.returncode != 0:
176 |             logger.warning(f"Command failed with return code {process.returncode}: {command}")
177 |             logger.debug(f"Command error output: {stderr_str}")
178 | 
179 |             if is_auth_error(stderr_str):
180 |                 return CommandResult(status="error", output=f"Authentication error: {stderr_str}\nPlease check your AWS credentials.")
181 | 
182 |             return CommandResult(status="error", output=stderr_str or "Command failed with no error output")
183 | 
184 |         return CommandResult(status="success", output=stdout_str)
185 |     except asyncio.CancelledError:
186 |         raise
187 |     except Exception as e:
188 |         raise CommandExecutionError(f"Failed to execute command: {str(e)}") from e
189 | 
190 | 
191 | async def execute_pipe_command(pipe_command: str, timeout: int | None = None) -> CommandResult:
192 |     """Execute a command that contains pipes.
193 | 
194 |     Validates and executes a piped command where output is fed into subsequent commands.
195 |     The first command must be an AWS CLI command, and subsequent commands must be
196 |     allowed Unix utilities.
197 | 
198 |     Args:
199 |         pipe_command: The piped command to execute
200 |         timeout: Optional timeout in seconds (defaults to DEFAULT_TIMEOUT)
201 | 
202 |     Returns:
203 |         CommandResult containing output and status
204 | 
205 |     Raises:
206 |         CommandValidationError: If any command in the pipe is invalid
207 |         CommandExecutionError: If the command fails to execute
208 |     """
209 |     # Validate the pipe command
210 |     try:
211 |         validate_pipe_command(pipe_command)
212 |     except ValueError as e:
213 |         raise CommandValidationError(f"Invalid pipe command: {str(e)}") from e
214 |     except CommandValidationError as e:
215 |         raise CommandValidationError(f"Invalid pipe command: {str(e)}") from e
216 | 
217 |     # Check if the first command in the pipe is an EC2 command and needs a region
218 |     from aws_mcp_server.config import AWS_REGION
219 | 
220 |     commands = split_pipe_command(pipe_command)
221 |     if commands:
222 |         # Split first command by spaces to check for EC2 service specifically
223 |         first_cmd_parts = shlex.split(commands[0])
224 |         is_ec2_command = len(first_cmd_parts) >= 2 and first_cmd_parts[0] == "aws" and first_cmd_parts[1] == "ec2"
225 |         has_region = "--region" in first_cmd_parts
226 | 
227 |         if is_ec2_command and not has_region:
228 |             # Add the region parameter to the first command
229 |             commands[0] = f"{commands[0]} --region {AWS_REGION}"
230 |             # Rebuild the pipe command
231 |             pipe_command = " | ".join(commands)
232 |             logger.debug(f"Added region to piped command: {pipe_command}")
233 | 
234 |     logger.debug(f"Executing piped command: {pipe_command}")
235 | 
236 |     try:
237 |         # Execute the piped command using our tools module
238 |         return await execute_piped_command(pipe_command, timeout)
239 |     except Exception as e:
240 |         raise CommandExecutionError(f"Failed to execute piped command: {str(e)}") from e
241 | 
242 | 
243 | async def get_command_help(service: str, command: str | None = None) -> CommandHelpResult:
244 |     """Get help documentation for an AWS CLI service or command.
245 | 
246 |     Retrieves the help documentation for a specified AWS service or command
247 |     by executing the appropriate AWS CLI help command.
248 | 
249 |     Args:
250 |         service: The AWS service (e.g., s3, ec2)
251 |         command: Optional command within the service
252 | 
253 |     Returns:
254 |         CommandHelpResult containing the help text
255 | 
256 |     Raises:
257 |         CommandExecutionError: If the help command fails
258 |     """
259 |     # Build the help command
260 |     cmd_parts: list[str] = ["aws", service]
261 |     if command:
262 |         cmd_parts.append(command)
263 |     cmd_parts.append("help")
264 | 
265 |     cmd_str = " ".join(cmd_parts)
266 | 
267 |     try:
268 |         logger.debug(f"Getting command help for: {cmd_str}")
269 |         result = await execute_aws_command(cmd_str)
270 | 
271 |         help_text = result["output"] if result["status"] == "success" else f"Error: {result['output']}"
272 | 
273 |         return CommandHelpResult(help_text=help_text)
274 |     except CommandValidationError as e:
275 |         logger.warning(f"Command validation error while getting help: {e}")
276 |         return CommandHelpResult(help_text=f"Command validation error: {str(e)}")
277 |     except CommandExecutionError as e:
278 |         logger.warning(f"Command execution error while getting help: {e}")
279 |         return CommandHelpResult(help_text=f"Error retrieving help: {str(e)}")
280 |     except Exception as e:
281 |         logger.error(f"Unexpected error while getting command help: {e}", exc_info=True)
282 |         return CommandHelpResult(help_text=f"Error retrieving help: {str(e)}")
283 | 
```

--------------------------------------------------------------------------------
/tests/unit/test_tools.py:
--------------------------------------------------------------------------------

```python
  1 | """Unit tests for the tools module."""
  2 | 
  3 | import asyncio
  4 | from unittest.mock import AsyncMock, MagicMock, patch
  5 | 
  6 | import pytest
  7 | 
  8 | from aws_mcp_server.tools import (
  9 |     ALLOWED_UNIX_COMMANDS,
 10 |     execute_piped_command,
 11 |     is_pipe_command,
 12 |     split_pipe_command,
 13 |     validate_unix_command,
 14 | )
 15 | 
 16 | 
 17 | def test_allowed_unix_commands():
 18 |     """Test that ALLOWED_UNIX_COMMANDS contains expected commands."""
 19 |     # Verify that common Unix utilities are in the allowed list
 20 |     common_commands = ["grep", "xargs", "cat", "ls", "wc", "sort", "uniq", "jq"]
 21 |     for cmd in common_commands:
 22 |         assert cmd in ALLOWED_UNIX_COMMANDS
 23 | 
 24 | 
 25 | def test_validate_unix_command():
 26 |     """Test the validate_unix_command function."""
 27 |     # Test valid commands
 28 |     for cmd in ["grep pattern", "ls -la", "wc -l", "cat file.txt"]:
 29 |         assert validate_unix_command(cmd), f"Command should be valid: {cmd}"
 30 | 
 31 |     # Test invalid commands
 32 |     for cmd in ["invalid_cmd", "sudo ls", ""]:
 33 |         assert not validate_unix_command(cmd), f"Command should be invalid: {cmd}"
 34 | 
 35 | 
 36 | def test_is_pipe_command():
 37 |     """Test the is_pipe_command function."""
 38 |     # Test commands with pipes
 39 |     assert is_pipe_command("aws s3 ls | grep bucket")
 40 |     assert is_pipe_command("aws s3api list-buckets | jq '.Buckets[].Name' | sort")
 41 | 
 42 |     # Test commands without pipes
 43 |     assert not is_pipe_command("aws s3 ls")
 44 |     assert not is_pipe_command("aws ec2 describe-instances")
 45 | 
 46 |     # Test commands with pipes in quotes (should not be detected as pipe commands)
 47 |     assert not is_pipe_command("aws s3 ls 's3://my-bucket/file|other'")
 48 |     assert not is_pipe_command('aws ec2 run-instances --user-data "echo hello | grep world"')
 49 | 
 50 |     # Test commands with escaped quotes - these should not confuse the parser
 51 |     assert is_pipe_command('aws s3 ls --query "Name=\\"value\\"" | grep bucket')
 52 |     assert not is_pipe_command('aws s3 ls "s3://my-bucket/file\\"|other"')
 53 | 
 54 | 
 55 | def test_split_pipe_command():
 56 |     """Test the split_pipe_command function."""
 57 |     # Test simple pipe command
 58 |     cmd = "aws s3 ls | grep bucket"
 59 |     result = split_pipe_command(cmd)
 60 |     assert result == ["aws s3 ls", "grep bucket"]
 61 | 
 62 |     # Test multi-pipe command
 63 |     cmd = "aws s3api list-buckets | jq '.Buckets[].Name' | sort"
 64 |     result = split_pipe_command(cmd)
 65 |     assert result == ["aws s3api list-buckets", "jq '.Buckets[].Name'", "sort"]
 66 | 
 67 |     # Test with quoted pipe symbols (should not split inside quotes)
 68 |     cmd = "aws s3 ls 's3://bucket/file|name' | grep 'pattern|other'"
 69 |     result = split_pipe_command(cmd)
 70 |     assert result == ["aws s3 ls 's3://bucket/file|name'", "grep 'pattern|other'"]
 71 | 
 72 |     # Test with double quotes
 73 |     cmd = 'aws s3 ls "s3://bucket/file|name" | grep "pattern|other"'
 74 |     result = split_pipe_command(cmd)
 75 |     assert result == ['aws s3 ls "s3://bucket/file|name"', 'grep "pattern|other"']
 76 | 
 77 |     # Test with escaped quotes
 78 |     cmd = 'aws s3 ls --query "Name=\\"value\\"" | grep bucket'
 79 |     result = split_pipe_command(cmd)
 80 |     assert result == ['aws s3 ls --query "Name=\\"value\\""', "grep bucket"]
 81 | 
 82 |     # Test with escaped pipe symbol in quotes
 83 |     cmd = 'aws s3 ls "s3://bucket/file\\"|name" | grep pattern'
 84 |     result = split_pipe_command(cmd)
 85 |     assert result == ['aws s3 ls "s3://bucket/file\\"|name"', "grep pattern"]
 86 | 
 87 | 
 88 | @pytest.mark.asyncio
 89 | async def test_execute_piped_command_success():
 90 |     """Test successful execution of a piped command."""
 91 |     with patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_subprocess:
 92 |         # Mock the first process in the pipe
 93 |         first_process_mock = AsyncMock()
 94 |         first_process_mock.returncode = 0
 95 |         first_process_mock.communicate.return_value = (b"S3 output", b"")
 96 | 
 97 |         # Mock the second process in the pipe
 98 |         second_process_mock = AsyncMock()
 99 |         second_process_mock.returncode = 0
100 |         second_process_mock.communicate.return_value = (b"Filtered output", b"")
101 | 
102 |         # Set up the mock to return different values on subsequent calls
103 |         mock_subprocess.side_effect = [first_process_mock, second_process_mock]
104 | 
105 |         result = await execute_piped_command("aws s3 ls | grep bucket")
106 | 
107 |         assert result["status"] == "success"
108 |         assert result["output"] == "Filtered output"
109 | 
110 |         # Verify first command was called with correct args
111 |         mock_subprocess.assert_any_call("aws", "s3", "ls", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
112 | 
113 |         # Verify second command was called with correct args
114 |         mock_subprocess.assert_any_call("grep", "bucket", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
115 | 
116 | 
117 | @pytest.mark.asyncio
118 | async def test_execute_piped_command_error_first_command():
119 |     """Test error handling in execute_piped_command when first command fails."""
120 |     with patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_subprocess:
121 |         # Mock a failed first process
122 |         process_mock = AsyncMock()
123 |         process_mock.returncode = 1
124 |         process_mock.communicate.return_value = (b"", b"Command failed: aws")
125 |         mock_subprocess.return_value = process_mock
126 | 
127 |         result = await execute_piped_command("aws s3 ls | grep bucket")
128 | 
129 |         assert result["status"] == "error"
130 |         assert "Command failed: aws" in result["output"]
131 | 
132 | 
133 | @pytest.mark.asyncio
134 | async def test_execute_piped_command_error_second_command():
135 |     """Test error handling in execute_piped_command when second command fails."""
136 |     with patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_subprocess:
137 |         # Mock the first process in the pipe (success)
138 |         first_process_mock = AsyncMock()
139 |         first_process_mock.returncode = 0
140 |         first_process_mock.communicate.return_value = (b"S3 output", b"")
141 | 
142 |         # Mock the second process in the pipe (failure)
143 |         second_process_mock = AsyncMock()
144 |         second_process_mock.returncode = 1
145 |         second_process_mock.communicate.return_value = (b"", b"Command not found: xyz")
146 | 
147 |         # Set up the mock to return different values on subsequent calls
148 |         mock_subprocess.side_effect = [first_process_mock, second_process_mock]
149 | 
150 |         result = await execute_piped_command("aws s3 ls | xyz")
151 | 
152 |         assert result["status"] == "error"
153 |         assert "Command not found: xyz" in result["output"]
154 | 
155 | 
156 | @pytest.mark.asyncio
157 | async def test_execute_piped_command_timeout():
158 |     """Test timeout handling in execute_piped_command."""
159 |     with patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_subprocess:
160 |         # Mock a process that times out
161 |         process_mock = AsyncMock()
162 |         # Use a properly awaitable mock that raises TimeoutError
163 |         communicate_mock = AsyncMock(side_effect=asyncio.TimeoutError())
164 |         process_mock.communicate = communicate_mock
165 |         # Use regular MagicMock since kill() is not an async method
166 |         process_mock.kill = MagicMock()
167 |         mock_subprocess.return_value = process_mock
168 | 
169 |         result = await execute_piped_command("aws s3 ls | grep bucket", timeout=1)
170 | 
171 |         assert result["status"] == "error"
172 |         assert "Command timed out after 1 seconds" in result["output"]
173 |         process_mock.kill.assert_called_once()
174 | 
175 | 
176 | @pytest.mark.asyncio
177 | async def test_execute_piped_command_exception():
178 |     """Test general exception handling in execute_piped_command."""
179 |     with patch("asyncio.create_subprocess_exec", side_effect=Exception("Test exception")):
180 |         result = await execute_piped_command("aws s3 ls | grep bucket")
181 | 
182 |         assert result["status"] == "error"
183 |         assert "Failed to execute command" in result["output"]
184 |         assert "Test exception" in result["output"]
185 | 
186 | 
187 | @pytest.mark.asyncio
188 | async def test_execute_piped_command_empty_command():
189 |     """Test handling of empty commands."""
190 |     result = await execute_piped_command("")
191 | 
192 |     assert result["status"] == "error"
193 |     assert "Empty command" in result["output"]
194 | 
195 | 
196 | @pytest.mark.asyncio
197 | async def test_execute_piped_command_timeout_during_final_wait():
198 |     """Test timeout handling during wait for the final command in a pipe."""
199 |     # This test directly tests the branch where a timeout occurs during awaiting the final command
200 |     with patch("asyncio.wait_for", side_effect=asyncio.TimeoutError()):
201 |         with patch("aws_mcp_server.tools.split_pipe_command") as mock_split:
202 |             mock_split.return_value = ["aws s3 ls", "grep bucket"]
203 | 
204 |             # We don't need to mock the subprocess - it won't reach that point
205 |             # because wait_for will raise a TimeoutError first
206 |             result = await execute_piped_command("aws s3 ls | grep bucket", timeout=5)
207 | 
208 |             assert result["status"] == "error"
209 |             assert "Command timed out after 5 seconds" in result["output"]
210 | 
211 | 
212 | @pytest.mark.asyncio
213 | async def test_execute_piped_command_kill_error_during_timeout():
214 |     """Test error handling when killing a process after timeout fails."""
215 |     with patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_subprocess:
216 |         # Mock a process that times out
217 |         process_mock = AsyncMock()
218 |         process_mock.communicate.side_effect = asyncio.TimeoutError()
219 |         process_mock.kill = MagicMock(side_effect=Exception("Failed to kill process"))
220 |         mock_subprocess.return_value = process_mock
221 | 
222 |         result = await execute_piped_command("aws s3 ls", timeout=1)
223 | 
224 |         assert result["status"] == "error"
225 |         assert "Command timed out after 1 seconds" in result["output"]
226 |         process_mock.kill.assert_called_once()
227 | 
228 | 
229 | @pytest.mark.asyncio
230 | async def test_execute_piped_command_large_output():
231 |     """Test output truncation in execute_piped_command."""
232 |     from aws_mcp_server.config import MAX_OUTPUT_SIZE
233 | 
234 |     with patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_subprocess:
235 |         # Mock a process with large output
236 |         process_mock = AsyncMock()
237 |         process_mock.returncode = 0
238 | 
239 |         # Generate output larger than MAX_OUTPUT_SIZE
240 |         large_output = "x" * (MAX_OUTPUT_SIZE + 1000)
241 |         process_mock.communicate.return_value = (large_output.encode("utf-8"), b"")
242 |         mock_subprocess.return_value = process_mock
243 | 
244 |         result = await execute_piped_command("aws s3 ls")
245 | 
246 |         assert result["status"] == "success"
247 |         assert len(result["output"]) <= MAX_OUTPUT_SIZE + 100  # Allow for truncation message
248 |         assert "output truncated" in result["output"]
249 | 
```

--------------------------------------------------------------------------------
/tests/integration/test_aws_live.py:
--------------------------------------------------------------------------------

```python
  1 | """Live AWS integration tests for the AWS MCP Server.
  2 | 
  3 | These tests connect to real AWS resources and require:
  4 | 1. AWS CLI installed locally
  5 | 2. AWS credentials configured with access to test resources
  6 | 3. The --run-integration flag when running pytest
  7 | 
  8 | Note: The tests that require an S3 bucket will create a temporary bucket
  9 | if AWS_TEST_BUCKET environment variable is not set.
 10 | """
 11 | 
 12 | import asyncio
 13 | import json
 14 | import logging
 15 | import os
 16 | import time
 17 | import uuid
 18 | 
 19 | import pytest
 20 | 
 21 | from aws_mcp_server.server import aws_cli_help, aws_cli_pipeline
 22 | 
 23 | # Configure logging
 24 | logging.basicConfig(level=logging.INFO)
 25 | logger = logging.getLogger(__name__)
 26 | 
 27 | 
 28 | class TestAWSLiveIntegration:
 29 |     """Integration tests that interact with real AWS services.
 30 | 
 31 |     These tests require AWS credentials and actual AWS resources.
 32 |     They verify the AWS MCP Server can properly interact with AWS.
 33 |     """
 34 | 
 35 |     # Apply the integration marker to each test method instead of the class
 36 | 
 37 |     @pytest.mark.asyncio
 38 |     @pytest.mark.integration
 39 |     @pytest.mark.parametrize(
 40 |         "service,command,expected_content",
 41 |         [
 42 |             ("s3", None, ["description", "ls", "cp", "mv"]),
 43 |             ("ec2", None, ["description", "run-instances", "describe-instances"]),
 44 |             # The AWS CLI outputs help with control characters that complicate exact matching
 45 |             # We need to use content that will be in the help text even with the escape characters
 46 |             ("s3", "ls", ["list s3 objects", "options", "examples"]),
 47 |         ],
 48 |     )
 49 |     async def test_aws_cli_help(self, ensure_aws_credentials, service, command, expected_content):
 50 |         """Test getting help for various AWS commands."""
 51 |         result = await aws_cli_help(service=service, command=command, ctx=None)
 52 | 
 53 |         # Verify we got a valid response
 54 |         assert isinstance(result, dict)
 55 |         assert "help_text" in result
 56 | 
 57 |         # Check for expected content in the help text (case-insensitive)
 58 |         help_text = result["help_text"].lower()
 59 |         for content in expected_content:
 60 |             assert content.lower() in help_text, f"Expected '{content}' in {service} {command} help text"
 61 | 
 62 |     @pytest.mark.asyncio
 63 |     @pytest.mark.integration
 64 |     async def test_list_s3_buckets(self, ensure_aws_credentials):
 65 |         """Test listing S3 buckets."""
 66 |         result = await aws_cli_pipeline(command="aws s3 ls", timeout=None, ctx=None)
 67 | 
 68 |         # Verify the result format
 69 |         assert isinstance(result, dict)
 70 |         assert "status" in result
 71 |         assert "output" in result
 72 |         assert result["status"] == "success"
 73 | 
 74 |         # Output should be a string containing the bucket listing (or empty if no buckets)
 75 |         assert isinstance(result["output"], str)
 76 | 
 77 |         logger.info(f"S3 bucket list result: {result['output']}")
 78 | 
 79 |     @pytest.mark.asyncio
 80 |     @pytest.mark.integration
 81 |     async def test_s3_operations_with_test_bucket(self, ensure_aws_credentials):
 82 |         """Test S3 operations using a test bucket.
 83 | 
 84 |         This test:
 85 |         1. Creates a temporary bucket
 86 |         2. Creates a test file
 87 |         3. Uploads it to S3
 88 |         4. Lists the bucket contents
 89 |         5. Downloads the file with a different name
 90 |         6. Verifies the downloaded content
 91 |         7. Cleans up all test files and the bucket
 92 |         """
 93 |         # Get region from environment or use default
 94 |         region = os.environ.get("AWS_TEST_REGION", os.environ.get("AWS_REGION", "us-east-1"))
 95 |         print(f"Using AWS region: {region}")
 96 | 
 97 |         # Generate a unique bucket name
 98 |         timestamp = int(time.time())
 99 |         random_id = str(uuid.uuid4())[:8]
100 |         bucket_name = f"aws-mcp-test-{timestamp}-{random_id}"
101 | 
102 |         test_file_name = "test_file.txt"
103 |         test_file_content = "This is a test file for AWS MCP Server integration tests"
104 |         downloaded_file_name = "test_file_downloaded.txt"
105 | 
106 |         try:
107 |             # Create the bucket
108 |             create_cmd = f"aws s3 mb s3://{bucket_name} --region {region}"
109 |             result = await aws_cli_pipeline(command=create_cmd, timeout=None, ctx=None)
110 |             assert result["status"] == "success", f"Failed to create bucket: {result['output']}"
111 | 
112 |             # Wait for bucket to be fully available
113 |             await asyncio.sleep(3)
114 | 
115 |             # Create a local test file
116 |             with open(test_file_name, "w") as f:
117 |                 f.write(test_file_content)
118 | 
119 |             # Upload the file to S3
120 |             upload_result = await aws_cli_pipeline(
121 |                 command=f"aws s3 cp {test_file_name} s3://{bucket_name}/{test_file_name} --region {region}", timeout=None, ctx=None
122 |             )
123 |             assert upload_result["status"] == "success"
124 | 
125 |             # List the bucket contents
126 |             list_result = await aws_cli_pipeline(command=f"aws s3 ls s3://{bucket_name}/ --region {region}", timeout=None, ctx=None)
127 |             assert list_result["status"] == "success"
128 |             assert test_file_name in list_result["output"]
129 | 
130 |             # Download the file with a different name
131 |             download_result = await aws_cli_pipeline(
132 |                 command=f"aws s3 cp s3://{bucket_name}/{test_file_name} {downloaded_file_name} --region {region}", timeout=None, ctx=None
133 |             )
134 |             assert download_result["status"] == "success"
135 | 
136 |             # Verify the downloaded file content
137 |             with open(downloaded_file_name, "r") as f:
138 |                 downloaded_content = f.read()
139 |             assert downloaded_content == test_file_content
140 | 
141 |         finally:
142 |             # Clean up local files
143 |             for file_name in [test_file_name, downloaded_file_name]:
144 |                 if os.path.exists(file_name):
145 |                     os.remove(file_name)
146 | 
147 |             # Clean up: Remove files from S3
148 |             await aws_cli_pipeline(command=f"aws s3 rm s3://{bucket_name} --recursive --region {region}", timeout=None, ctx=None)
149 | 
150 |             # Delete the bucket
151 |             await aws_cli_pipeline(command=f"aws s3 rb s3://{bucket_name} --region {region}", timeout=None, ctx=None)
152 | 
153 |     @pytest.mark.asyncio
154 |     @pytest.mark.integration
155 |     @pytest.mark.parametrize(
156 |         "command,expected_attributes,description",
157 |         [
158 |             # Test JSON formatting with EC2 regions
159 |             ("aws ec2 describe-regions --output json", {"json_key": "Regions", "expected_type": list}, "JSON output with EC2 regions"),
160 |             # Test JSON formatting with S3 buckets (may be empty but should be valid JSON)
161 |             ("aws s3api list-buckets --output json", {"json_key": "Buckets", "expected_type": list}, "JSON output with S3 buckets"),
162 |         ],
163 |     )
164 |     async def test_aws_json_output_formatting(self, ensure_aws_credentials, command, expected_attributes, description):
165 |         """Test JSON output formatting from various AWS commands."""
166 |         result = await aws_cli_pipeline(command=command, timeout=None, ctx=None)
167 | 
168 |         assert result["status"] == "success", f"Command failed: {result.get('output', '')}"
169 | 
170 |         # The output should be valid JSON
171 |         try:
172 |             json_data = json.loads(result["output"])
173 | 
174 |             # Verify expected JSON structure
175 |             json_key = expected_attributes["json_key"]
176 |             expected_type = expected_attributes["expected_type"]
177 | 
178 |             assert json_key in json_data, f"Expected key '{json_key}' not found in JSON response"
179 |             assert isinstance(json_data[json_key], expected_type), f"Expected {json_key} to be of type {expected_type.__name__}"
180 | 
181 |             # Log some info about the response
182 |             logger.info(f"Successfully parsed JSON response for {description} with {len(json_data[json_key])} items")
183 | 
184 |         except json.JSONDecodeError:
185 |             pytest.fail(f"Output is not valid JSON: {result['output'][:100]}...")
186 | 
187 |     @pytest.mark.asyncio
188 |     @pytest.mark.integration
189 |     @pytest.mark.parametrize(
190 |         "command,validation_func,description",
191 |         [
192 |             # Test simple pipe with count
193 |             ("aws ec2 describe-regions --query 'Regions[*].RegionName' --output text | wc -l", lambda output: int(output.strip()) > 0, "Count of AWS regions"),
194 |             # Test pipe with grep and sort
195 |             (
196 |                 "aws ec2 describe-regions --query 'Regions[*].RegionName' --output text | grep east | sort",
197 |                 lambda output: all("east" in r.lower() for r in output.strip().split("\n") if r),
198 |                 "Filtered and sorted east regions",
199 |             ),
200 |             # Test more complex pipe with multiple operations
201 |             (
202 |                 "aws ec2 describe-regions --output json | grep RegionName | head -3 | wc -l",
203 |                 lambda output: int(output.strip()) <= 3,
204 |                 "Limited region output with multiple pipes",
205 |             ),
206 |             # Test pipe with JSON grep
207 |             (
208 |                 "aws iam list-roles --output json | grep RoleName",
209 |                 lambda output: "RoleName" in output or output.strip() == "",
210 |                 "Lists IAM roles or returns empty if none exist",
211 |             ),
212 |             # Very simple pipe command that should work anywhere
213 |             (
214 |                 "aws --version | grep aws",
215 |                 lambda output: "aws" in output.lower(),  # Just check for the word "aws" in output
216 |                 "AWS version with grep",
217 |             ),
218 |         ],
219 |     )
220 |     async def test_piped_commands(self, ensure_aws_credentials, command, validation_func, description):
221 |         """Test execution of various piped commands with AWS CLI and Unix utilities."""
222 |         result = await aws_cli_pipeline(command=command, timeout=None, ctx=None)
223 | 
224 |         assert result["status"] == "success", f"Command failed: {result.get('output', '')}"
225 | 
226 |         # Validate the output using the provided validation function
227 |         assert validation_func(result["output"]), f"Output validation failed for {description}"
228 | 
229 |         # Log success
230 |         logger.info(f"Successfully executed piped command for {description}: {result['output'][:50]}...")
231 | 
232 |     @pytest.mark.asyncio
233 |     @pytest.mark.integration
234 |     async def test_aws_account_resource(self, ensure_aws_credentials):
235 |         """Test that the AWS account resource returns non-null account information."""
236 |         # Import resources module
237 |         from aws_mcp_server.resources import get_aws_account_info
238 | 
239 |         # Get account info directly using the function
240 |         account_info = get_aws_account_info()
241 | 
242 |         # Verify account info is not empty
243 |         assert account_info is not None, "AWS account info is None"
244 | 
245 |         # Verify the account_id field is not null
246 |         # We don't check specific values, just that they are not null when credentials are present
247 |         assert account_info["account_id"] is not None, "AWS account_id is null"
248 | 
249 |         # Log success with masked account ID for verification (show first 4 chars)
250 |         account_id = account_info["account_id"]
251 |         masked_id = f"{account_id[:4]}{'*' * (len(account_id) - 4)}" if account_id else "None"
252 |         logger.info(f"Successfully accessed AWS account info with account_id: {masked_id}")
253 | 
254 |         # Log organization_id status - this might be null depending on permissions
255 |         has_org_id = account_info["organization_id"] is not None
256 |         logger.info(f"Organization ID available: {has_org_id}")
257 | 
258 |     @pytest.mark.asyncio
259 |     @pytest.mark.integration
260 |     async def test_us_east_1_region_services(self, ensure_aws_credentials):
261 |         """Test that the us-east-1 region resource returns expected services.
262 | 
263 |         This test verifies that:
264 |         1. The region details endpoint for us-east-1 works
265 |         2. The core AWS services we expect are listed as available
266 |         3. The service information is correctly formatted
267 |         """
268 |         # Import resources module and server
269 |         from aws_mcp_server.resources import get_region_details
270 |         from aws_mcp_server.server import mcp
271 | 
272 |         # Get region details directly using the function
273 |         region_code = "us-east-1"
274 |         region_details = get_region_details(region_code)
275 | 
276 |         # Verify region details is not empty
277 |         assert region_details is not None, "Region details is None"
278 |         assert region_details["code"] == region_code, "Region code does not match expected value"
279 |         assert region_details["name"] == "US East (N. Virginia)", "Region name does not match expected value"
280 | 
281 |         # Verify services is a list and not empty
282 |         assert "services" in region_details, "Services not found in region details"
283 |         assert isinstance(region_details["services"], list), "Services is not a list"
284 |         assert len(region_details["services"]) > 0, "Services list is empty"
285 | 
286 |         # Verify each service has id and name fields
287 |         for service in region_details["services"]:
288 |             assert "id" in service, "Service missing 'id' field"
289 |             assert "name" in service, "Service missing 'name' field"
290 | 
291 |         # Check for core AWS services that should be available in us-east-1
292 |         required_services = ["ec2", "s3", "lambda", "dynamodb", "rds", "cloudformation", "iam"]
293 | 
294 |         service_ids = [service["id"] for service in region_details["services"]]
295 | 
296 |         for required_service in required_services:
297 |             assert required_service in service_ids, f"Required service '{required_service}' not found in us-east-1 services"
298 | 
299 |         # Log the number of services found
300 |         logger.info(f"Found {len(region_details['services'])} services in us-east-1")
301 | 
302 |         # Test access through the MCP resource URI
303 |         try:
304 |             resource = await mcp.resources_read(uri=f"aws://config/regions/{region_code}")
305 |             assert resource is not None, "Failed to read region resource through MCP"
306 |             assert resource.content["code"] == region_code, "Resource region code does not match"
307 |             assert resource.content["name"] == "US East (N. Virginia)", "Resource region name does not match"
308 |             assert "services" in resource.content, "Services not found in MCP resource content"
309 | 
310 |             # Verify at least the same core services are present in the resource response
311 |             mcp_service_ids = [service["id"] for service in resource.content["services"]]
312 |             for required_service in required_services:
313 |                 assert required_service in mcp_service_ids, f"Required service '{required_service}' not found in MCP resource services"
314 | 
315 |             logger.info("Successfully accessed us-east-1 region details through MCP resource")
316 |         except Exception as e:
317 |             logger.warning(f"Could not test MCP resource access: {e}")
318 |             # Don't fail the test if this part doesn't work - focus on the direct API test
319 | 
```

