DUMMY PR TO TEST - Add python.yml and fix related code issues.

.github/workflows/python.yml

@@ -0,0 +1,181 @@
+name: Python
+
+on:
+  push:
+  pull_request:
+  workflow_dispatch:
+
+permissions: {}
+
+jobs:
+  build:
+    strategy:
+      fail-fast: false
+      matrix:
+        job_id: ["Build"]
+    defaults:
+      run:
+        working-directory: .
+    name: Build AWS MCP Proxy
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      pull-requests: read
+      security-events: write
+      actions: read
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v6.3.1
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version-file: ".python-version"
+          # cache: uv (not supported)
+
+      - name: Install dependencies
+        run: uv sync --frozen --all-extras --dev
+
+      - name: Run tests
+        run: |
+          if [ -d "tests" ]; then
+            uv run --frozen pytest --cov --cov-branch --cov-report=term-missing --cov-report=xml:${{ matrix.package }}-coverage.xml
+          else
+            echo "No tests directory found, skipping tests"
+          fi
+
+#      - name: Upload coverage reports to Codecov
+#        uses: codecov/codecov-action@18283e04ce6e62d37312384ff67231eb8fd56d24 #v5.4.3
+#        with:
+#          token: ${{ secrets.CODECOV_TOKEN }}
+#          files: ${{ matrix.package }}-coverage.xml
+
+      - name: Run pyright
+        run: uv run --frozen pyright
+
+      - name: Run ruff format
+        run: uv run --frozen ruff format .
+
+      - name: Run ruff check
+        run: uv run --frozen ruff check .
+
+      - name: Build package
+        run: uv build
+
+      - name: Upload distribution
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+        with:
+          path: dist/
+
+      - name: Generate Software Bill of Materials (SBOM)
+        run: |
+          source .venv/bin/activate
+          echo "Attempt to convert to proper UTF-8 files https://github.com/CycloneDX/cyclonedx-python/issues/868"
+          find .venv -type f -path '*/*.dist-info/*' > .venv/FILES
+          # because grep with xargs returns 123 have to do this the long and hard way...
+          while IFS= read -r line; do
+            (grep -s -q -axv '.*' $line &&
+              if [[ "$(file -b --mime-encoding $line)" != "binary" ]]; then
+                echo "illegal utf-8 characters in $line...converting...";
+                iconv -f $(file -b --mime-encoding $line) -t utf-8 $line > $line.utf8;
+                mv $line.utf8 $line;
+              fi;
+            ) || echo "good $line"
+          done < .venv/FILES;
+          uv tool run --from cyclonedx-bom==6.1.3 cyclonedx-py environment $VIRTUAL_ENV --PEP-639 --gather-license-texts --pyproject pyproject.toml --mc-type library --output-format JSON > sbom.json
+      - name: Display SBOM
+        run: |
+          cat <<EOT |
+          import re
+          import json
+          import importlib.metadata as metadata
+
+          def parse_bom(json_file):
+              # Parse the JSON file
+              with open(json_file, 'r') as file:
+                  data = json.load(file)
+
+              # Extract components
+              components = []
+              for component in data['components']:
+                  comp_info = {}
+
+                  # Get name, version, description, and purl
+                  comp_info['name'] = component.get('name', 'Unknown')
+                  comp_info['version'] = component.get('version', 'Unknown')
+                  comp_info['description'] = component.get('description', 'Unknown')
+                  comp_info['purl'] = component.get('purl', 'Unknown')
+
+                  # Get licenses
+                  comp_info['licenses'] = []
+                  licenses = component.get('licenses', [])
+                  for license in licenses:
+                      if license.get('license', {}).get('id'):
+                          comp_info['licenses'].append(license.get('license').get('id'))
+                  if len(comp_info['licenses']) == 0:
+                      comp_info['licenses'].append("No licenses")
+
+                  # Extract additional information (copyright, etc.)
+                  copyright_info = extract_copyright_from_metadata(comp_info['name'])
+                  comp_info['copyright'] = copyright_info if copyright_info else "No copyright information"
+
+                  components.append(comp_info)
+
+              return components
+
+          def extract_copyright_from_metadata(package_name):
+              try:
+                  # Use importlib.metadata to retrieve metadata from the installed package
+                  dist = metadata.distribution(package_name)
+                  metadata_info = dist.metadata
+
+                  # Extract relevant metadata
+                  copyright_info = []
+                  author = metadata_info.get('Author')
+                  author_email = metadata_info.get('Author-email')
+                  license_info = metadata_info.get('License')
+
+                  if author:
+                      copyright_info.append(f"Author: {author}")
+                  if author_email:
+                      copyright_info.append(f"Author Email: {author_email}")
+                  if license_info:
+                      copyright_info.append(f"License: {license_info}")
+
+                  # Check for classifiers or any extra metadata fields
+                  if 'Classifier' in metadata_info:
+                      for classifier in metadata_info.get_all('Classifier'):
+                          if 'copyright' in classifier.lower():
+                              copyright_info.append(classifier)
+
+                  return ', '.join(copyright_info) if copyright_info else None
+
+              except metadata.PackageNotFoundError:
+                  return None
+
+
+          def main():
+              bom_file = 'sbom.json'  # Replace with your BOM file path
+              components = parse_bom(bom_file)
+
+              for component in components:
+                  print(f"Name: {component['name']}")
+                  print(f"Version: {component['version']}")
+                  print(f"Description: {component['description']}")
+                  print(f"PURL: {component['purl']}")
+                  print(f"Licenses: {', '.join(component['licenses'])}")
+                  print(f"Copyright: {component['copyright']}")
+                  print("-" * 40)
+
+          if __name__ == "__main__":
+              main()
+          EOT
+           python -
+
+      - name: Upload Software Bill of Materials
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+        with:
+          name: sbom-aws-mcp-proxy
+          path: sbom.json

src/aws_mcp_proxy/server.py

@@ -24,14 +24,14 @@
 import argparse
 import asyncio
 import os
+from fastmcp.server.server import FastMCP
+from loguru import logger
 from src.aws_mcp_proxy.mcp_proxy_manager import McpProxyManager
 from src.aws_mcp_proxy.utils import (
     create_transport_with_sigv4,
     determine_service_name,
     normalize_endpoint_url,
 )
-from fastmcp.server.server import FastMCP
-from loguru import logger
 from typing import Any
 
 

src/aws_mcp_proxy/utils.py

@@ -13,11 +13,11 @@
 # limitations under the License.
 
 """Utility functions for the AWS MCP Proxy."""
-import logging
 import re
-from src.aws_mcp_proxy.sigv4_helper import create_sigv4_client
 from fastmcp.client.transports import StreamableHttpTransport
-from typing import Optional
+from httpx import AsyncClient, Auth, Timeout
+from src.aws_mcp_proxy.sigv4_helper import create_sigv4_client
+from typing import Dict, Optional
 from urllib.parse import urlparse
 
 
@@ -34,11 +34,22 @@ def create_transport_with_sigv4(
     Returns:
         StreamableHttpTransport instance with SigV4 authentication
     """
+    def client_factory(
+        headers: Optional[Dict[str, str]] = None,
+        timeout: Optional[Timeout] = None,
+        auth: Optional[Auth] = None
+    ) -> AsyncClient:
+        return create_sigv4_client(
+            service=service,
+            profile=profile,
+            headers=headers,
+            timeout=timeout,
+            auth=auth
+        )
+
     return StreamableHttpTransport(
         url=url,
-        httpx_client_factory=lambda **kwargs: create_sigv4_client(
-            service=service, profile=profile, **kwargs
-        ),
+        httpx_client_factory=client_factory,
     )
 
 

tests/test_mcp_proxy_manager.py

@@ -15,8 +15,8 @@
 """Tests for mcp_proxy_manager module."""
 
 import pytest
-from src.aws_mcp_proxy.mcp_proxy_manager import McpProxyManager
 from fastmcp.server.server import FastMCP
+from src.aws_mcp_proxy.mcp_proxy_manager import McpProxyManager
 from unittest.mock import AsyncMock, MagicMock
 
 

tests/test_utils.py

@@ -15,12 +15,13 @@
 """Tests for utils module."""
 
 import pytest
+from fastmcp.client.transports import StreamableHttpTransport
+from httpx import Timeout
 from src.aws_mcp_proxy.utils import (
     create_transport_with_sigv4,
     determine_service_name,
     normalize_endpoint_url,
 )
-from fastmcp.client.transports import StreamableHttpTransport
 from unittest.mock import MagicMock, patch
 
 
@@ -47,11 +48,16 @@ def test_create_transport_with_sigv4(self, mock_create_sigv4_client):
         # We need to access the factory through the transport's internal structure
         if hasattr(result, 'httpx_client_factory') and result.httpx_client_factory:
             factory = result.httpx_client_factory
-            test_kwargs = {'timeout': 30}
-            factory(**test_kwargs)
+            test_timeout = Timeout(30.0)
+            test_headers = {'Content-Type': 'application/json'}
+            factory(headers=test_headers, timeout=test_timeout)
 
             mock_create_sigv4_client.assert_called_once_with(
-                service=service, profile=profile, timeout=30
+                service=service,
+                profile=profile,
+                headers=test_headers,
+                timeout=test_timeout,
+                auth=None
             )
         else:
             # If we can't access the factory directly, just verify the transport was created
@@ -71,7 +77,13 @@ def test_create_transport_with_sigv4_no_profile(self, mock_create_sigv4_client):
             factory = result.httpx_client_factory
             factory()
 
-            mock_create_sigv4_client.assert_called_once_with(service=service, profile=None)
+            mock_create_sigv4_client.assert_called_once_with(
+                service=service,
+                profile=None,
+                headers=None,
+                timeout=None,
+                auth=None
+            )
         else:
             # If we can't access the factory directly, just verify the transport was created
             assert result is not None