feat: simplify header filtering to use simple allowlist

- Remove CustomHeadersConfig class complexity
- Change custom_headers to header_allowlist as simple list[str] | None
- Remove unnecessary strategy field (allowlist is the only option)
- Update all service methods and tests to use simplified structure
- Much cleaner manifest configuration:

  # Old
  custom_headers:
    strategy: allowlist
    allowed_headers: [...]

  # New
  header_allowlist: [...]

Breaking change: Agent manifest header configuration simplified

Author: prassanna-ravishankar

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

@@ -5,7 +5,7 @@
 from agentex.lib.core.tracing.tracer import AsyncTracer
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.utils.temporal import heartbeat_if_in_workflow
-from agentex.lib.types.agent_configs import CustomHeadersConfig
+# No longer need CustomHeadersConfig import
 from agentex.types.event import Event
 from agentex.types.task import Task
 from agentex.types.task_message import TaskMessage
@@ -24,64 +24,64 @@ def __init__(
         self._agentex_client = agentex_client
         self._tracer = tracer
 
-    async def _get_agent_header_config(
+    async def _get_agent_header_allowlist(
         self, agent_name: str | None = None, agent_id: str | None = None
-    ) -> CustomHeadersConfig | None:
+    ) -> list[str] | None:
         """
-        Get agent's header configuration from manifest.
+        Get agent's header allowlist from manifest.
         
         Returns None for pass-through behavior (all headers forwarded).
-        Returns CustomHeadersConfig only if agent has specific filtering requirements.
+        Returns list[str] of allowed header patterns if agent has filtering requirements.
         
         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
+        based on agent_name/agent_id and return the header_allowlist. For now, we implement
         pass-through by default as requested.
         
-        TODO: Integrate with agent registry to load manifest-based header configurations
+        TODO: Integrate with agent registry to load manifest-based header allowlists
         """
         # 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
+        # 3. Return manifest.agent.header_allowlist
         return None
 
     def _filter_headers(
         self,
         headers: dict[str, str] | None,
-        config: CustomHeadersConfig | None
+        allowlist: list[str] | None
     ) -> dict[str, str]:
         """
-        Filter headers based on agent's configuration.
+        Filter headers based on agent's allowlist.
         
-        Pass-through by default: if no config provided, all headers are forwarded.
-        If config exists, only allowlisted headers are forwarded.
+        Pass-through by default: if no allowlist provided, all headers are forwarded.
+        If allowlist exists, only matching headers are forwarded.
         
         Args:
             headers: Headers to filter
-            config: Agent's header configuration (None = pass-through all)
+            allowlist: Agent's header allowlist (None = pass-through all)
             
         Returns:
             Filtered headers dictionary
         """
         if not headers:
             return {}
 
-        # Pass-through behavior: if no config, forward all headers
-        if config is None:
-            logger.info("No header filtering config found, passing through all %d headers", len(headers))
+        # Pass-through behavior: if no allowlist, forward all headers
+        if allowlist is None:
+            logger.info("No header allowlist found, passing through all %d headers", len(headers))
             return headers
 
         # Apply filtering based on allowlist
-        if not config.allowed_headers:
-            logger.info("Empty allowlist in config, blocking all headers")
+        if not allowlist:
+            logger.info("Empty allowlist, blocking all headers")
             return {}
 
         filtered = {}
         for header_name, header_value in headers.items():
             # Check against allowlist patterns (case-insensitive)
             header_allowed = False
-            for pattern in config.allowed_headers:
+            for pattern in allowlist:
                 if fnmatch.fnmatch(header_name.lower(), pattern.lower()):
                     header_allowed = True
                     break
@@ -120,8 +120,8 @@ async def task_create(
 
             # Extract headers from request and filter them
             headers = request.get("headers") if request else None
-            header_config = await self._get_agent_header_config(agent_name, agent_id)
-            filtered_headers = self._filter_headers(headers, header_config)
+            header_allowlist = await self._get_agent_header_allowlist(agent_name, agent_id)
+            filtered_headers = self._filter_headers(headers, header_allowlist)
 
             if agent_name:
                 json_rpc_response = await self._agentex_client.agents.rpc_by_name(
@@ -178,8 +178,8 @@ async def message_send(
 
             # Extract headers from request and filter them
             headers = request.get("headers") if request else None
-            header_config = await self._get_agent_header_config(agent_name, agent_id)
-            filtered_headers = self._filter_headers(headers, header_config)
+            header_allowlist = await self._get_agent_header_allowlist(agent_name, agent_id)
+            filtered_headers = self._filter_headers(headers, header_allowlist)
 
             if agent_name:
                 json_rpc_response = await self._agentex_client.agents.rpc_by_name(
@@ -245,8 +245,8 @@ async def event_send(
 
             # Extract headers from request and filter them
             headers = request.get("headers") if request else None
-            header_config = await self._get_agent_header_config(agent_name, agent_id)
-            filtered_headers = self._filter_headers(headers, header_config)
+            header_allowlist = await self._get_agent_header_allowlist(agent_name, agent_id)
+            filtered_headers = self._filter_headers(headers, header_allowlist)
 
             if agent_name:
                 json_rpc_response = await self._agentex_client.agents.rpc_by_name(
@@ -327,8 +327,8 @@ async def task_cancel(
 
             # Extract headers from request and filter them
             headers = request.get("headers") if request else None
-            header_config = await self._get_agent_header_config(agent_name, agent_id)
-            filtered_headers = self._filter_headers(headers, header_config)
+            header_allowlist = await self._get_agent_header_allowlist(agent_name, agent_id)
+            filtered_headers = self._filter_headers(headers, header_allowlist)
 
             # Build params for the agent (task identification)
             params = {}

src/agentex/lib/sdk/config/agent_config.py

@@ -2,7 +2,7 @@
 
 from pydantic import Field
 
-from agentex.lib.types.agent_configs import TemporalConfig, TemporalWorkflowConfig, CustomHeadersConfig
+from agentex.lib.types.agent_configs import TemporalConfig, TemporalWorkflowConfig
 from agentex.lib.types.credentials import CredentialMapping
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.utils.model_utils import BaseModel
@@ -28,8 +28,8 @@ class AgentConfig(BaseModel):
     temporal: TemporalConfig | None = Field(
         default=None, description="Temporal workflow configuration for this agent"
     )
-    custom_headers: CustomHeadersConfig | None = Field(
-        default=None, description="Configuration for custom HTTP headers that this agent can receive"
+    header_allowlist: list[str] | None = Field(
+        default=None, description="List of header patterns this agent is allowed to receive. If not provided, all headers are forwarded."
     )
 
     def is_temporal_agent(self) -> bool:

src/agentex/lib/types/agent_configs.py

@@ -1,4 +1,3 @@
-from typing import Literal
 from pydantic import BaseModel, Field, field_validator, model_validator
 
 
@@ -38,26 +37,7 @@ class TemporalWorkerConfig(BaseModel):
     )
 
 
-class CustomHeadersConfig(BaseModel):
-    """
-    Configuration for custom HTTP header filtering for agents.
-    
-    When present in agent configuration, only allowlisted headers are forwarded.
-    When absent, all headers are passed through (pass-through by default).
-    
-    Attributes:
-        strategy: Header filtering strategy (currently only 'allowlist' supported)
-        allowed_headers: List of header patterns that are allowed to be forwarded
-    """
-    
-    strategy: Literal["allowlist"] = Field(
-        default="allowlist",
-        description="Header filtering strategy. Only 'allowlist' is currently supported."
-    )
-    allowed_headers: list[str] = Field(
-        default_factory=list,
-        description="List of header name patterns that are allowed to be forwarded to this agent. Supports wildcards with fnmatch."
-    )
+# Removed CustomHeadersConfig - now using simple list[str] for header_allowlist
 
 
 class TemporalConfig(BaseModel):

tests/test_header_filtering.py

@@ -4,50 +4,46 @@
 import fnmatch
 import pytest
 
-from agentex.lib.types.agent_configs import CustomHeadersConfig
+# No longer need CustomHeadersConfig - using simple lists now
 
 
-class TestCustomHeadersConfig:
-    """Test CustomHeadersConfig functionality."""
+class TestHeaderAllowlist:
+    """Test header allowlist functionality."""
 
-    def test_custom_headers_config_creation(self) -> None:
-        """Test that CustomHeadersConfig can be created with defaults."""
-        config = CustomHeadersConfig()
-        
-        assert config.strategy == "allowlist"
-        assert config.allowed_headers == []
-
-    def test_custom_headers_config_with_headers(self) -> None:
-        """Test CustomHeadersConfig with allowed headers."""
-        config = CustomHeadersConfig(
-            allowed_headers=["x-user-email", "x-tenant-id"]
-        )
-        
-        assert config.strategy == "allowlist"
-        assert config.allowed_headers == ["x-user-email", "x-tenant-id"]
+    def test_allowlist_creation(self) -> None:
+        """Test that allowlists can be created as simple lists."""
+        allowlist = ["x-user-email", "x-tenant-id"]
+        assert len(allowlist) == 2
+        assert "x-user-email" in allowlist
+        assert "x-tenant-id" in allowlist
+
+    def test_empty_allowlist(self) -> None:
+        """Test empty allowlist behavior."""
+        allowlist: list[str] = []
+        assert len(allowlist) == 0
 
 
 def filter_headers_standalone(
     headers: dict[str, str] | None,
-    config: CustomHeadersConfig | None
+    allowlist: list[str] | None
 ) -> dict[str, str]:
     """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:
+    # Pass-through behavior: if no allowlist, forward all headers
+    if allowlist is None:
         return headers
 
     # Apply filtering based on allowlist
-    if not config.allowed_headers:
+    if not allowlist:
         return {}
 
     filtered = {}
     for header_name, header_value in headers.items():
         # Check against allowlist patterns (case-insensitive)
         header_allowed = False
-        for pattern in config.allowed_headers:
+        for pattern in allowlist:
             if fnmatch.fnmatch(header_name.lower(), pattern.lower()):
                 header_allowed = True
                 break
@@ -63,42 +59,42 @@ class TestHeaderFiltering:
 
     def test_filter_headers_no_headers(self) -> None:
         """Test header filtering with no input headers."""
-        config = CustomHeadersConfig(allowed_headers=["x-user-email"])
-        result = filter_headers_standalone(None, config)
+        allowlist = ["x-user-email"]
+        result = filter_headers_standalone(None, allowlist)
         assert result == {}
 
-        result = filter_headers_standalone({}, config)
+        result = filter_headers_standalone({}, allowlist)
         assert result == {}
 
     def test_filter_headers_pass_through_by_default(self) -> None:
-        """Test header filtering with no config (pass-through behavior)."""
+        """Test header filtering with no allowlist (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
+        # All headers should pass through when no allowlist is provided
         assert result == headers
 
     def test_filter_headers_empty_allowlist(self) -> None:
-        """Test header filtering with empty allowed headers."""
-        config = CustomHeadersConfig(allowed_headers=[])
+        """Test header filtering with empty allowlist."""
+        allowlist: list[str] = []
         headers = {"x-user-email": "[email protected]", "x-admin-token": "secret"}
-        result = filter_headers_standalone(headers, config)
+        result = filter_headers_standalone(headers, allowlist)
         assert result == {}
 
     def test_filter_headers_allowed_headers(self) -> None:
-        """Test header filtering with allowed headers."""
-        config = CustomHeadersConfig(allowed_headers=["x-user-email", "x-tenant-id"])
+        """Test header filtering with allowlist."""
+        allowlist = ["x-user-email", "x-tenant-id"]
         headers = {
             "x-user-email": "[email protected]",
             "x-tenant-id": "tenant123",
             "x-admin-token": "secret",  # Should be filtered out
             "content-type": "application/json"  # Should be filtered out
         }
-        result = filter_headers_standalone(headers, config)
+        result = filter_headers_standalone(headers, allowlist)
 
         expected = {
             "x-user-email": "[email protected]",
@@ -108,14 +104,14 @@ def test_filter_headers_allowed_headers(self) -> None:
 
     def test_filter_headers_case_insensitive_patterns(self) -> None:
         """Test header filtering with case-insensitive pattern matching."""
-        config = CustomHeadersConfig(allowed_headers=["X-User-Email", "x-tenant-*"])
+        allowlist = ["X-User-Email", "x-tenant-*"]
         headers = {
             "x-user-email": "[email protected]",  # Should match X-User-Email
             "X-TENANT-ID": "tenant123",  # Should match x-tenant-*
             "x-tenant-name": "acme",  # Should match x-tenant-*
             "x-admin-token": "secret"  # Should be filtered out
         }
-        result = filter_headers_standalone(headers, config)
+        result = filter_headers_standalone(headers, allowlist)
 
         expected = {
             "x-user-email": "[email protected]",
@@ -126,7 +122,7 @@ def test_filter_headers_case_insensitive_patterns(self) -> None:
 
     def test_filter_headers_wildcard_patterns(self) -> None:
         """Test header filtering with wildcard patterns."""
-        config = CustomHeadersConfig(allowed_headers=["x-user-*", "authorization"])
+        allowlist = ["x-user-*", "authorization"]
         headers = {
             "x-user-id": "123",
             "x-user-email": "[email protected]", 
@@ -135,7 +131,7 @@ def test_filter_headers_wildcard_patterns(self) -> None:
             "x-system-info": "blocked",  # Should be filtered out
             "content-type": "application/json"  # Should be filtered out
         }
-        result = filter_headers_standalone(headers, config)
+        result = filter_headers_standalone(headers, allowlist)
 
         expected = {
             "x-user-id": "123",
@@ -147,7 +143,7 @@ def test_filter_headers_wildcard_patterns(self) -> None:
 
     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*"])
+        allowlist = ["x-tenant-*", "x-user-[abc]*", "auth*"]
         headers = {
             "x-tenant-id": "tenant1",     # Matches x-tenant-*
             "x-tenant-name": "acme",      # Matches x-tenant-*  
@@ -158,7 +154,7 @@ def test_filter_headers_complex_patterns(self) -> None:
             "authenticate": "digest",     # Matches auth*
             "content-type": "json",       # No match
         }
-        result = filter_headers_standalone(headers, config)
+        result = filter_headers_standalone(headers, allowlist)
 
         expected = {
             "x-tenant-id": "tenant1",
@@ -172,7 +168,7 @@ def test_filter_headers_complex_patterns(self) -> None:
 
     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-*"])
+        allowlist = ["authorization", "accept-language", "custom-*"]
         headers = {
             "authorization": "Bearer token",
             "accept-language": "en-US",
@@ -181,7 +177,7 @@ def test_filter_headers_all_types(self) -> None:
             "content-type": "application/json",  # Should be blocked
             "x-blocked": "value"  # Should be blocked
         }
-        result = filter_headers_standalone(headers, config)
+        result = filter_headers_standalone(headers, allowlist)
 
         expected = {
             "authorization": "Bearer token",