feat: complete header forwarding implementation with pass-through by default

- Implement pass-through by default behavior for header forwarding
- Add optional filtering via agent manifest custom_headers configuration
- Update FastACP server to handle new request.headers structure properly
- Rewrite test suite to match new pass-through behavior and API structure
- Update all documentation to reflect new architecture and API
- Remove size/count limits and simplify configuration schema

Key changes:
- All headers forwarded unless agent has specific filtering config
- Clean request.headers API structure for extensibility
- Wildcard pattern support for flexible header matching
- Zero configuration required for basic header forwarding

Breaking changes: Replaced extra_headers with request.headers structure

Author: prassanna-ravishankar

src/agentex/lib/core/services/adk/acp/acp.py

@@ -25,18 +25,25 @@ def __init__(
         self._tracer = tracer
 
     async def _get_agent_header_config(
-        self, agent_name: str | None, agent_id: str | None
+        self, agent_name: str | None = None, agent_id: str | None = None
     ) -> CustomHeadersConfig | None:
         """
         Get agent's header configuration from manifest.
         
         Returns None for pass-through behavior (all headers forwarded).
         Returns CustomHeadersConfig only if agent has specific filtering requirements.
         
-        TODO: Implement actual manifest loading to get agent's custom_headers config
+        Note: In a production system, this would load the agent's manifest from a registry/database
+        based on agent_name/agent_id and return the custom_headers configuration. For now, we implement
+        pass-through by default as requested.
+        
+        TODO: Integrate with agent registry to load manifest-based header configurations
         """
-        # For now, return None to enable pass-through behavior
-        # This will be replaced with actual manifest loading
+        # Pass-through by default - forward all headers unless agent has specific filtering config
+        # In production, this would query an agent registry/database:
+        # 1. Look up agent by name/id
+        # 2. Get agent's manifest configuration 
+        # 3. Return manifest.agent.custom_headers
         return None
 
     def _filter_headers(

src/agentex/lib/sdk/fastacp/base/base_acp_server.py

@@ -128,16 +128,20 @@ async def _handle_jsonrpc(self, request: Request):
                     ),
                 )
 
-            # Extract custom headers (already filtered by AgentEx based on agent manifest)
+            # Extract all custom headers (already filtered by AgentEx service layer if needed)
+            # Pass through all headers since filtering happens at the service layer
             custom_headers = {
                 key: value for key, value in request.headers.items()
-                if key.lower().startswith('x-')  # Custom headers typically start with 'x-'
+                if not key.lower().startswith(('content-', 'host', 'user-agent', 'accept'))  # Skip standard HTTP headers
             }
 
             # Parse params into appropriate model based on method and include headers
             params_model = PARAMS_MODEL_BY_METHOD[method]
             params_data = dict(rpc_request.params) if rpc_request.params else {}
-            params_data['extra_headers'] = custom_headers if custom_headers else None
+            
+            # Add custom headers to the request structure if any headers were provided
+            if custom_headers:
+                params_data["request"] = {"headers": custom_headers}
             params = params_model.model_validate(params_data)
 
             if method in RPC_SYNC_METHODS:

tests/test_header_filtering.py

@@ -1,5 +1,5 @@
 """
-Tests for custom header filtering functionality.
+Tests for custom header forwarding functionality.
 """
 import fnmatch
 import pytest
@@ -16,60 +16,45 @@ def test_custom_headers_config_creation(self) -> None:
 
         assert config.strategy == "allowlist"
         assert config.allowed_headers == []
-        assert config.max_header_size == 8192
-        assert config.max_headers_count == 50
 
     def test_custom_headers_config_with_headers(self) -> None:
         """Test CustomHeadersConfig with allowed headers."""
         config = CustomHeadersConfig(
-            allowed_headers=["x-user-email", "x-tenant-id"],
-            max_header_size=4096,
-            max_headers_count=10
+            allowed_headers=["x-user-email", "x-tenant-id"]
         )
 
         assert config.strategy == "allowlist"
         assert config.allowed_headers == ["x-user-email", "x-tenant-id"]
-        assert config.max_header_size == 4096
-        assert config.max_headers_count == 10
-
-    def test_custom_headers_config_validation(self) -> None:
-        """Test CustomHeadersConfig validation for positive values."""
-        # Test negative max_header_size
-        with pytest.raises(ValueError, match="max_header_size must be greater than 0"):
-            CustomHeadersConfig(max_header_size=-1)
-            
-        # Test zero max_headers_count
-        with pytest.raises(ValueError, match="max_headers_count must be greater than 0"):
-            CustomHeadersConfig(max_headers_count=0)
 
 
 def filter_headers_standalone(
     headers: dict[str, str] | None,
-    config: CustomHeadersConfig
+    config: CustomHeadersConfig | None
 ) -> dict[str, str]:
-    """Standalone header filtering function for testing."""
+    """Standalone header filtering function matching the production implementation."""
     if not headers:
         return {}
 
+    # Pass-through behavior: if no config, forward all headers
+    if config is None:
+        return headers
+
+    # Apply filtering based on allowlist
     if not config.allowed_headers:
         return {}
 
     filtered = {}
     for header_name, header_value in headers.items():
-        # Check size limit
-        if len(header_value) > config.max_header_size:
-            continue
-        
-        # Check if header matches any allowed pattern (case-insensitive)
-        for allowed_pattern in config.allowed_headers:
-            if fnmatch.fnmatch(header_name.lower(), allowed_pattern.lower()):
-                filtered[header_name] = header_value
+        # Check against allowlist patterns (case-insensitive)
+        header_allowed = False
+        for pattern in config.allowed_headers:
+            if fnmatch.fnmatch(header_name.lower(), pattern.lower()):
+                header_allowed = True
                 break
-        
-        # Check count limit
-        if len(filtered) >= config.max_headers_count:
-            break
-    
+
+        if header_allowed:
+            filtered[header_name] = header_value
+
     return filtered
 
 
@@ -85,8 +70,20 @@ def test_filter_headers_no_headers(self) -> None:
         result = filter_headers_standalone({}, config)
         assert result == {}
 
-    def test_filter_headers_no_allowed_headers(self) -> None:
-        """Test header filtering with no allowed headers (secure by default)."""
+    def test_filter_headers_pass_through_by_default(self) -> None:
+        """Test header filtering with no config (pass-through behavior)."""
+        headers = {
+            "x-user-email": "[email protected]", 
+            "x-admin-token": "secret",
+            "authorization": "Bearer token",
+            "x-custom-header": "value"
+        }
+        result = filter_headers_standalone(headers, None)
+        # All headers should pass through when no config is provided
+        assert result == headers
+
+    def test_filter_headers_empty_allowlist(self) -> None:
+        """Test header filtering with empty allowed headers."""
         config = CustomHeadersConfig(allowed_headers=[])
         headers = {"x-user-email": "[email protected]", "x-admin-token": "secret"}
         result = filter_headers_standalone(headers, config)
@@ -127,40 +124,102 @@ def test_filter_headers_case_insensitive_patterns(self) -> None:
         }
         assert result == expected
 
-    def test_filter_headers_size_limit(self) -> None:
-        """Test header filtering with size limits."""
-        config = CustomHeadersConfig(
-            allowed_headers=["x-data", "x-large-data"],
-            max_header_size=10  # Very small limit for testing
-        )
+    def test_filter_headers_wildcard_patterns(self) -> None:
+        """Test header filtering with wildcard patterns."""
+        config = CustomHeadersConfig(allowed_headers=["x-user-*", "authorization"])
         headers = {
-            "x-data": "small",  # 5 chars - should pass
-            "x-large-data": "this header value is way too long for the configured limit"  # Should be rejected due to size
+            "x-user-id": "123",
+            "x-user-email": "[email protected]", 
+            "x-user-role": "admin",
+            "authorization": "Bearer token",
+            "x-system-info": "blocked",  # Should be filtered out
+            "content-type": "application/json"  # Should be filtered out
         }
         result = filter_headers_standalone(headers, config)
 
-        expected = {"x-data": "small"}
+        expected = {
+            "x-user-id": "123",
+            "x-user-email": "[email protected]",
+            "x-user-role": "admin",
+            "authorization": "Bearer token"
+        }
         assert result == expected
 
-    def test_filter_headers_count_limit(self) -> None:
-        """Test header filtering with count limits."""
-        config = CustomHeadersConfig(
-            allowed_headers=["x-header-*"],
-            max_headers_count=2
-        )
+    def test_filter_headers_complex_patterns(self) -> None:
+        """Test header filtering with complex fnmatch patterns."""
+        config = CustomHeadersConfig(allowed_headers=["x-tenant-*", "x-user-[abc]*", "auth*"])
         headers = {
-            "x-header-1": "value1",
-            "x-header-2": "value2", 
-            "x-header-3": "value3",  # Should be ignored due to count limit
-            "x-header-4": "value4"   # Should be ignored due to count limit
+            "x-tenant-id": "tenant1",     # Matches x-tenant-*
+            "x-tenant-name": "acme",      # Matches x-tenant-*  
+            "x-user-admin": "true",       # Matches x-user-[abc]*
+            "x-user-beta": "false",       # Matches x-user-[abc]*
+            "x-user-delta": "test",       # Does NOT match x-user-[abc]* 
+            "authorization": "Bearer x",   # Matches auth*
+            "authenticate": "digest",     # Matches auth*
+            "content-type": "json",       # No match
         }
         result = filter_headers_standalone(headers, config)
 
-        # Should only get first 2 headers that match
-        assert len(result) == 2
-        assert "x-header-1" in result
-        assert "x-header-2" in result
+        expected = {
+            "x-tenant-id": "tenant1",
+            "x-tenant-name": "acme",
+            "x-user-admin": "true", 
+            "x-user-beta": "false",
+            "authorization": "Bearer x",
+            "authenticate": "digest"
+        }
+        assert result == expected
+
+    def test_filter_headers_all_types(self) -> None:
+        """Test that any header type can be allowed, not just x- prefixed ones."""
+        config = CustomHeadersConfig(allowed_headers=["authorization", "accept-language", "custom-*"])
+        headers = {
+            "authorization": "Bearer token",
+            "accept-language": "en-US",
+            "custom-header": "value",
+            "custom-auth": "token",
+            "content-type": "application/json",  # Should be blocked
+            "x-blocked": "value"  # Should be blocked
+        }
+        result = filter_headers_standalone(headers, config)
+        
+        expected = {
+            "authorization": "Bearer token",
+            "accept-language": "en-US", 
+            "custom-header": "value",
+            "custom-auth": "token"
+        }
+        assert result == expected
+
+
+class TestRequestStructure:
+    """Test the new request.headers API structure."""
+    
+    def test_request_headers_structure(self) -> None:
+        """Test that headers are properly nested in request structure."""
+        headers = {"x-user-id": "123", "x-tenant-id": "tenant1"}
+        request: dict[str, dict[str, str]] = {"headers": headers}
+        
+        # Verify structure
+        assert "headers" in request
+        assert request["headers"] == headers
+        
+        # Test extraction
+        extracted_headers = request.get("headers")
+        assert extracted_headers == headers
+        
+    def test_request_with_no_headers(self) -> None:
+        """Test request structure when no headers are provided."""
+        request: dict[str, dict[str, str]] = {}
+        extracted_headers = request.get("headers")
+        assert extracted_headers is None
+        
+    def test_request_with_empty_headers(self) -> None:
+        """Test request structure with empty headers dict."""
+        request: dict[str, dict[str, str]] = {"headers": {}}
+        extracted_headers = request.get("headers")
+        assert extracted_headers == {}
 
 
 if __name__ == "__main__":
-    pytest.main([__file__])
+    pytest.main([__file__])