feat(server.py): add forwarding region as optional argument

Author: Kyon Caldera

mcp_proxy_for_aws/cli.py

@@ -60,10 +60,16 @@ def parse_args():
 
     parser.add_argument(
         '--region',
-        help='AWS region to use (uses AWS_REGION environment variable if not provided, with final fallback to us-east-1)',
+        help='AWS region to sign (uses AWS_REGION environment variable if not provided, with final fallback to us-east-1)',
         default=None,
     )
 
+    parser.add_argument(
+        '--forwarding-region',
+        help='AWS region to forward to server (uses --region if not provided)',
+        default=None,
+    )
+    
     parser.add_argument(
         '--read-only',
         action='store_true',

mcp_proxy_for_aws/server.py

@@ -52,13 +52,20 @@ async def setup_mcp_mode(local_mcp: FastMCP, args) -> None:
 
     # Validate and determine region
     region = determine_aws_region(args.endpoint, args.region)
+    forwarding_region = args.forwarding_region or region
     logger.debug('Using region: %s', region)
 
     # Get profile
     profile = args.profile
 
     # Log server configuration
-    logger.info('Using service: %s, region: %s, profile: %s', service, region, profile)
+    logger.info(
+        'Using service: %s, region: %s, forwarding region: %s, profile: %s',
+        service,
+        region,
+        forwarding_region,
+        profile,
+    )
     logger.info('Running in MCP mode')
 
     timeout = httpx.Timeout(
@@ -69,7 +76,9 @@ async def setup_mcp_mode(local_mcp: FastMCP, args) -> None:
     )
 
     # Create transport with SigV4 authentication
-    transport = create_transport_with_sigv4(args.endpoint, service, region, timeout, profile)
+    transport = create_transport_with_sigv4(
+        args.endpoint, service, region, forwarding_region, timeout, profile
+    )
 
     # Create proxy with the transport
     proxy = FastMCP.as_proxy(transport)

mcp_proxy_for_aws/utils.py

@@ -32,6 +32,7 @@ def create_transport_with_sigv4(
     url: str,
     service: str,
     region: str,
+    forwarding_region: str,
     custom_timeout: httpx.Timeout,
     profile: Optional[str] = None,
 ) -> StreamableHttpTransport:
@@ -41,6 +42,7 @@ def create_transport_with_sigv4(
         url: The endpoint URL
         service: AWS service name for SigV4 signing
         region: AWS region to use
+        forwarding_region: AWS region to forward to server
         custom_timeout: httpx.Timeout used to connect to the endpoint
         profile: AWS profile to use (optional)
 
@@ -60,7 +62,7 @@ def client_factory(
             region=region,
             headers=headers,
             timeout=custom_timeout,
-            metadata={'AWS_REGION': region},
+            metadata={'AWS_REGION': forwarding_region},
             auth=auth,
         )
 

tests/integ/test_proxy_simple_mcp_server.py

@@ -124,3 +124,7 @@ async def test_metadata_injection_aws_region(
     assert 'AWS_REGION' in response_data['received_meta'], (
         f'Metadata should contain AWS_REGION: {response_data["received_meta"]}'
     )
+    assert (
+        response_data['received_meta']['AWS_REGION']
+        == remote_mcp_server_configuration['region_name']
+    ), f'AWS_REGION should be {remote_mcp_server_configuration["region_name"]}'

tests/unit/test_server.py

@@ -56,6 +56,7 @@ async def test_setup_mcp_mode(
         mock_args.profile = None
         mock_args.read_only = True
         mock_args.retries = 1
+        mock_args.forwarding_region = None
         # Add timeout parameters
         mock_args.timeout = 180.0
         mock_args.connect_timeout = 60.0
@@ -86,8 +87,9 @@ async def test_setup_mcp_mode(
         assert call_args[0][0] == 'https://test.example.com'
         assert call_args[0][1] == 'test-service'
         assert call_args[0][2] == 'us-east-1'
-        # call_args[0][3] is the Timeout object
-        assert call_args[0][4] is None  # profile
+        assert call_args[0][3] == 'us-east-1'  # forwarding_region (defaults to region)
+        # call_args[0][4] is the Timeout object
+        assert call_args[0][5] is None  # profile
         mock_as_proxy.assert_called_once_with(mock_transport)
         mock_add_filtering.assert_called_once_with(mock_proxy, True)
         mock_add_retry.assert_called_once_with(mock_proxy, 1)
@@ -116,6 +118,7 @@ async def test_setup_mcp_mode_no_retries(
         mock_args.profile = 'test-profile'
         mock_args.read_only = False
         mock_args.retries = 0  # No retries
+        mock_args.forwarding_region = 'eu-west-1'
         # Add timeout parameters
         mock_args.timeout = 180.0
         mock_args.connect_timeout = 60.0
@@ -146,8 +149,9 @@ async def test_setup_mcp_mode_no_retries(
         assert call_args[0][0] == 'https://test.example.com'
         assert call_args[0][1] == 'test-service'
         assert call_args[0][2] == 'us-east-1'
-        # call_args[0][3] is the Timeout object
-        assert call_args[0][4] == 'test-profile'  # profile
+        assert call_args[0][3] == 'eu-west-1'  # forwarding_region
+        # call_args[0][4] is the Timeout object
+        assert call_args[0][5] == 'test-profile'  # profile
         mock_as_proxy.assert_called_once_with(mock_transport)
         mock_add_filtering.assert_called_once_with(mock_proxy, False)
         mock_proxy.run_async.assert_called_once()

tests/unit/test_utils.py

@@ -39,9 +39,12 @@ def test_create_transport_with_sigv4(self, mock_create_sigv4_client):
         service = 'test-service'
         profile = 'test-profile'
         region = 'us-east-1'
+        forwarding_region = 'us-west-2'
         custom_timeout = Timeout(30.0)
 
-        result = create_transport_with_sigv4(url, service, region, custom_timeout, profile)
+        result = create_transport_with_sigv4(
+            url, service, region, forwarding_region, custom_timeout, profile
+        )
 
         # Verify result is StreamableHttpTransport
         assert isinstance(result, StreamableHttpTransport)
@@ -61,7 +64,7 @@ def test_create_transport_with_sigv4(self, mock_create_sigv4_client):
                 headers={'test': 'header'},
                 timeout=custom_timeout,
                 auth=None,
-                metadata={'AWS_REGION': region},
+                metadata={'AWS_REGION': forwarding_region},
             )
         else:
             # If we can't access the factory directly, just verify the transport was created
@@ -75,9 +78,12 @@ def test_create_transport_with_sigv4_no_profile(self, mock_create_sigv4_client):
         url = 'https://test-service.us-west-2.api.aws/mcp'
         service = 'test-service'
         region = 'test-region'
+        forwarding_region = 'test-forwarding-region'
         custom_timeout = Timeout(60.0)
 
-        result = create_transport_with_sigv4(url, service, region, custom_timeout)
+        result = create_transport_with_sigv4(
+            url, service, region, forwarding_region, custom_timeout
+        )
 
         # Test that the httpx_client_factory calls create_sigv4_client correctly
         # We need to access the factory through the transport's internal structure
@@ -92,7 +98,7 @@ def test_create_transport_with_sigv4_no_profile(self, mock_create_sigv4_client):
                 headers=None,
                 timeout=custom_timeout,
                 auth=None,
-                metadata={'AWS_REGION': region},
+                metadata={'AWS_REGION': forwarding_region},
             )
         else:
             # If we can't access the factory directly, just verify the transport was created