Cleanup: Removed dependencies, ran pre-commit with --run-all

Author: arangatang

.github/workflows/python.yml

@@ -178,4 +178,4 @@ jobs:
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
         with:
           name: sbom-aws-mcp-proxy
-          path: sbom.json
+          path: sbom.json

DEVELOPMENT.md

@@ -10,4 +10,4 @@ npx @modelcontextprotocol/inspector uv run \
 src/aws_mcp_proxy/server.py \
 --endpoint <your endpoint>
 ```
-Then click connect in the opened browser window.
+Then click connect in the opened browser window.

pyproject.toml

@@ -10,12 +10,9 @@ requires-python = ">=3.10"
 dependencies = [
     "fastmcp>=2.11.2",
     "mcp>=1.12.3",
-    "loguru>=0.7.0",
     "pydantic>=2.10.6",
     "boto3>=1.34.0",
     "botocore>=1.34.0",
-    "requests_auth_aws_sigv4",
-    "httpx-auth-awssigv4",
 ]
 license = {text = "Apache-2.0"}
 license-files = ["LICENSE", "NOTICE" ]

src/aws_mcp_proxy/logging_config.py

@@ -0,0 +1,36 @@
+"""Logging configuration for AWS MCP Proxy."""
+
+import logging
+import sys
+from typing import Optional
+
+
+def configure_logging(level: Optional[str] = None) -> None:
+    """Configure logging with a standard format and optional level.
+
+    Args:
+        level: Optional logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
+               If not provided, defaults to INFO.
+    """
+    # Set default level to INFO if not provided
+    log_level = getattr(logging, level.upper()) if level else logging.INFO
+
+    # Configure logging format
+    log_format = '%(asctime)s | %(levelname)s | %(name)s | %(message)s'
+    date_format = '%Y-%m-%d %H:%M:%S'
+
+    # Create console handler with formatting
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setFormatter(logging.Formatter(log_format, date_format))
+
+    # Configure root logger
+    root_logger = logging.getLogger()
+    root_logger.setLevel(log_level)
+
+    # Remove any existing handlers and add our console handler
+    root_logger.handlers.clear()
+    root_logger.addHandler(console_handler)
+
+    # Set httpx logging to WARNING by default to reduce noise
+    logging.getLogger('httpx').setLevel(logging.WARNING)
+    logging.getLogger('httpcore').setLevel(logging.WARNING)

src/aws_mcp_proxy/mcp_proxy_manager.py

@@ -14,13 +14,15 @@
 
 """MCP Proxy Manager for handling proxy content integration."""
 
+import logging
 from fastmcp.server.server import FastMCP
-from loguru import logger
 
 
 class McpProxyManager:
     """Manages the integration of proxy content (tools, resources, prompts) into MCP servers."""
 
+    logger = logging.getLogger(__name__)
+
     def __init__(self, target_mcp: FastMCP, allow_write: bool = False):
         """Initialize the MCP Proxy Manager.
 
@@ -45,10 +47,10 @@ async def add_proxy_content(self, proxy: FastMCP) -> None:
             await self._add_resources(proxy)
             await self._add_prompts(proxy)
 
-            logger.info('Successfully added proxy content to MCP server')
+            self.logger.info('Successfully added proxy content to MCP server')
 
         except Exception as e:
-            logger.error(f'Failed to add proxy content to MCP server: {e}')
+            self.logger.error(f'Failed to add proxy content to MCP server: {e}')
             raise
 
     async def _add_tools(self, proxy: FastMCP) -> None:
@@ -61,15 +63,15 @@ async def _add_tools(self, proxy: FastMCP) -> None:
             Exception: If tools cannot be retrieved or added
         """
         tools = await proxy.get_tools()
-        logger.info(f'Found {len(tools)} tools in proxy')
+        self.logger.info(f'Found {len(tools)} tools in proxy')
 
         for tool_name, tool in tools.items():
             # Check the tool annotations and disable if needed
             annotations = tool.annotations
             if not self.allow_write:
                 # In readOnly mode, skip the tools with no readOnlyHint=True annotation
                 if annotations and not annotations.readOnlyHint or not annotations:
-                    logger.info(f'Skipping tool {tool_name} needing write permissions')
+                    self.logger.info(f'Skipping tool {tool_name} needing write permissions')
                     continue
 
             self.target_mcp.add_tool(tool)
@@ -82,17 +84,17 @@ async def _add_resources(self, proxy: FastMCP) -> None:
         """
         try:
             resources = await proxy.get_resources()
-            logger.info(f'Found {len(resources)} resources in proxy')
+            self.logger.info(f'Found {len(resources)} resources in proxy')
 
             for resource_uri, resource in resources.items():
                 self.target_mcp.add_resource(resource)
-                logger.debug(f'Added resource: {resource_uri}')
+                self.logger.debug(f'Added resource: {resource_uri}')
 
         except AttributeError:
             # Proxy doesn't have resources, which is fine
-            logger.debug("Proxy doesn't have resources method")
+            self.logger.debug("Proxy doesn't have resources method")
         except Exception as e:
-            logger.warning(f'Failed to get resources from proxy: {e}')
+            self.logger.warning(f'Failed to get resources from proxy: {e}')
 
     async def _add_prompts(self, proxy: FastMCP) -> None:
         """Add prompts from proxy to target MCP server.
@@ -102,14 +104,14 @@ async def _add_prompts(self, proxy: FastMCP) -> None:
         """
         try:
             prompts = await proxy.get_prompts()
-            logger.info(f'Found {len(prompts)} prompts in proxy')
+            self.logger.info(f'Found {len(prompts)} prompts in proxy')
 
             for prompt_name, prompt in prompts.items():
                 self.target_mcp.add_prompt(prompt)
-                logger.debug(f'Added prompt: {prompt_name}')
+                self.logger.debug(f'Added prompt: {prompt_name}')
 
         except AttributeError:
             # Proxy doesn't have prompts, which is fine
-            logger.debug("Proxy doesn't have prompts method")
+            self.logger.debug("Proxy doesn't have prompts method")
         except Exception as e:
-            logger.warning(f'Failed to get prompts from proxy: {e}')
+            self.logger.warning(f'Failed to get prompts from proxy: {e}')

src/aws_mcp_proxy/server.py

@@ -23,9 +23,10 @@
 
 import argparse
 import asyncio
+import logging
 import os
 from fastmcp.server.server import FastMCP
-from loguru import logger
+from src.aws_mcp_proxy.logging_config import configure_logging
 from src.aws_mcp_proxy.mcp_proxy_manager import McpProxyManager
 from src.aws_mcp_proxy.utils import (
     create_transport_with_sigv4,
@@ -35,6 +36,9 @@
 from typing import Any
 
 
+logger = logging.getLogger(__name__)
+
+
 async def setup_mcp_mode(mcp: FastMCP, args) -> None:
     """Set up the server in MCP mode."""
     logger.info('Setting up server in MCP mode')
@@ -47,7 +51,9 @@ async def setup_mcp_mode(mcp: FastMCP, args) -> None:
     region = os.getenv('AWS_REGION', 'us-west-2')
 
     # Log server configuration
-    logger.info(f'Using service: {service}, region: {region}, profile: {profile or "default"}')
+    logger.info(
+        'Using service: %s, region: %s, profile: %s', service, region, profile or 'default'
+    )
     logger.info('Running in MCP mode')
 
     # Normalize endpoint URL
@@ -104,14 +110,22 @@ def parse_args():
         help='Allow tools that require write permissions to be enabled',
     )
 
+    parser.add_argument(
+        '--log-level',
+        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
+        default='INFO',
+        help='Set the logging level (default: INFO)',
+    )
+
     return parser.parse_args()
 
 
 def main():
     """Run the MCP server."""
     args = parse_args()
 
-    # Set up logging
+    # Configure logging
+    configure_logging(args.log_level)
     logger.info('Starting AWS MCP Proxy Server')
 
     # Create FastMCP instance
@@ -131,7 +145,7 @@ async def setup_and_run():
             await mcp.run_async()
 
         except Exception as e:
-            logger.error(f'Failed to start server: {e}')
+            logger.error('Failed to start server: %s', e)
             raise
 
     # Run the server

src/aws_mcp_proxy/sigv4_helper.py

@@ -16,10 +16,60 @@
 
 import boto3
 import httpx
+import logging
 import os
-from httpx_auth_awssigv4 import SigV4Auth
-from loguru import logger
-from typing import Optional
+from botocore.auth import SigV4Auth
+from botocore.awsrequest import AWSRequest
+from botocore.credentials import Credentials
+from typing import Any, Dict, Generator, Optional
+
+
+logger = logging.getLogger(__name__)
+
+
+class SigV4HTTPXAuth(httpx.Auth):
+    """HTTPX Auth class that signs requests with AWS SigV4."""
+
+    def __init__(
+        self,
+        credentials: Credentials,
+        service: str,
+        region: str,
+    ):
+        """Initialize SigV4HTTPXAuth.
+
+        Args:
+            credentials: AWS credentials to use for signing
+            service: AWS service name to sign requests for
+            region: AWS region to sign requests for
+        """
+        self.credentials = credentials
+        self.service = service
+        self.region = region
+        self.signer = SigV4Auth(credentials, service, region)
+
+    def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, None]:
+        """Signs the request with SigV4 and adds the signature to the request headers."""
+        # Create an AWS request
+        headers = dict(request.headers)
+        # Header 'connection' = 'keep-alive' is not used in calculating the request
+        # signature on the server-side, and results in a signature mismatch if included
+        headers.pop('connection', None)  # Remove if present, ignore if not
+
+        aws_request = AWSRequest(
+            method=request.method,
+            url=str(request.url),
+            data=request.content,
+            headers=headers,
+        )
+
+        # Sign the request with SigV4
+        self.signer.add_auth(aws_request)
+
+        # Add the signature header to the original request
+        request.headers.update(dict(aws_request.headers))
+
+        yield request
 
 
 async def _handle_error_response(response: httpx.Response) -> None:
@@ -39,24 +89,24 @@ async def _handle_error_response(response: httpx.Response) -> None:
             # Read response content to extract error details
             await response.aread()
         except Exception as e:
-            logger.error(f'Failed to read response: {e}')
+            logger.error('Failed to read response: %s', e)
 
         # Try to extract error details with fallbacks
         error_msg = ''
         try:
             # Try to parse JSON error details
             error_details = response.json()
-            logger.error(f'HTTP {response.status_code} Error Details: {error_details}')
+            logger.error('HTTP %d Error Details: %s', response.status_code, error_details)
             error_msg = f'HTTP {response.status_code}: {error_details} for url {response.url}'
         except Exception:
             # If JSON parsing fails, use response text or status code
             try:
                 response_text = response.text
-                logger.error(f'HTTP {response.status_code} Error: {response_text}')
+                logger.error('HTTP %d Error: %s', response.status_code, response_text)
                 error_msg = f'HTTP {response.status_code}: {response_text} for url {response.url}'
             except Exception:
                 # Fallback to just status code and URL
-                logger.error(f'HTTP {response.status_code} Error for url {response.url}')
+                logger.error('HTTP %d Error for url %s', response.status_code, response.url)
                 error_msg = f'HTTP {response.status_code} Error for url {response.url}'
 
         # Raise the status error with enhanced message
@@ -102,7 +152,7 @@ def create_aws_session(profile: Optional[str] = None) -> boto3.Session:
 
 def create_sigv4_auth(
     service: str, profile: Optional[str] = None, region: Optional[str] = None
-) -> SigV4Auth:
+) -> SigV4HTTPXAuth:
     """Create SigV4 authentication for AWS requests.
 
     Args:
@@ -111,7 +161,7 @@ def create_sigv4_auth(
         region: AWS region (defaults to AWS_REGION env var or us-west-2)
 
     Returns:
-        SigV4Auth instance
+        SigV4HTTPXAuth instance
 
     Raises:
         ValueError: If credentials cannot be obtained
@@ -125,25 +175,23 @@ def create_sigv4_auth(
         region = os.environ.get('AWS_REGION', 'us-west-2')
 
     # Create SigV4Auth with explicit credentials
-    sigv4_auth = SigV4Auth(
-        access_key=credentials.access_key,
-        secret_key=credentials.secret_key,
+    sigv4_auth = SigV4HTTPXAuth(
+        credentials=credentials,
         service=service,
         region=region,
-        token=credentials.token,
     )
 
-    logger.info(f"Created SigV4 authentication for service '{service}' in region '{region}'")
+    logger.info("Created SigV4 authentication for service '%s' in region '%s'", service, region)
     return sigv4_auth
 
 
 def create_sigv4_client(
     service: str = 'eks-mcp',
     profile: Optional[str] = None,
     region: Optional[str] = None,
-    headers=None,
-    auth=None,
-    **kwargs,
+    headers: Optional[Dict[str, str]] = None,
+    auth: Optional[httpx.Auth] = None,
+    **kwargs: Any,
 ) -> httpx.AsyncClient:
     """Create an httpx.AsyncClient with SigV4 authentication.
 
@@ -180,7 +228,7 @@ def create_sigv4_client(
     sigv4_auth = create_sigv4_auth(service, profile, region)
 
     # Create the client with SigV4 auth and error handling event hook
-    logger.info(f"Creating httpx.AsyncClient with SigV4 authentication for service '{service}'")
+    logger.info("Creating httpx.AsyncClient with SigV4 authentication for service '%s'", service)
 
     return httpx.AsyncClient(
         auth=sigv4_auth,