feat: restructure header forwarding API from extra_headers to request.headers

- Update CustomHeadersConfig schema to remove max_header_size and max_headers_count
- Change API from flat extra_headers parameter to nested request.headers structure
- Implement pass-through by default behavior (headers forwarded unless filtered)
- Update all ACP service methods, temporal activities, and ADK modules
- Remove size/count limits from header filtering logic

Breaking changes: extra_headers parameter replaced with request parameter across all ACP APIs

Author: prassanna-ravishankar

src/agentex/lib/adk/_modules/acp.py

@@ -59,7 +59,7 @@ async def create_task(
         start_to_close_timeout: timedelta = timedelta(seconds=5),
         heartbeat_timeout: timedelta = timedelta(seconds=5),
         retry_policy: RetryPolicy = DEFAULT_RETRY_POLICY,
-        extra_headers: dict[str, str] | None = None,
+        request: dict[str, Any] | None = None,
     ) -> Task:
         """
         Create a new task.
@@ -72,7 +72,7 @@ async def create_task(
             start_to_close_timeout: The start to close timeout for the task.
             heartbeat_timeout: The heartbeat timeout for the task.
             retry_policy: The retry policy for the task.
-            extra_headers: Additional HTTP headers to forward to the agent (filtered by agent's allowlist).
+            request: Additional request context including headers to forward to the agent.
 
         Returns:
             The task entry.
@@ -87,7 +87,7 @@ async def create_task(
                     params=params,
                     trace_id=trace_id,
                     parent_span_id=parent_span_id,
-                    extra_headers=extra_headers,
+                    request=request,
                 ),
                 response_type=Task,
                 start_to_close_timeout=start_to_close_timeout,
@@ -102,7 +102,7 @@ async def create_task(
                 params=params,
                 trace_id=trace_id,
                 parent_span_id=parent_span_id,
-                extra_headers=extra_headers,
+                request=request,
             )
 
     async def send_event(
@@ -116,7 +116,7 @@ async def send_event(
         start_to_close_timeout: timedelta = timedelta(seconds=5),
         heartbeat_timeout: timedelta = timedelta(seconds=5),
         retry_policy: RetryPolicy = DEFAULT_RETRY_POLICY,
-        extra_headers: dict[str, str] | None = None,
+        request: dict[str, Any] | None = None,
     ) -> Event:
         """
         Send an event to a task.
@@ -131,7 +131,7 @@ async def send_event(
             start_to_close_timeout: The start to close timeout for the event.
             heartbeat_timeout: The heartbeat timeout for the event.
             retry_policy: The retry policy for the event.
-            extra_headers: Additional HTTP headers to forward to the agent (filtered by agent's allowlist).
+            request: Additional request context including headers to forward to the agent.
 
         Returns:
             The event entry.
@@ -146,7 +146,7 @@ async def send_event(
                     content=content,
                     trace_id=trace_id,
                     parent_span_id=parent_span_id,
-                    extra_headers=extra_headers,
+                    request=request,
                 ),
                 response_type=None,
                 start_to_close_timeout=start_to_close_timeout,
@@ -161,7 +161,7 @@ async def send_event(
                 content=content,
                 trace_id=trace_id,
                 parent_span_id=parent_span_id,
-                extra_headers=extra_headers,
+                request=request,
             )
 
     async def send_message(
@@ -175,7 +175,7 @@ async def send_message(
         start_to_close_timeout: timedelta = timedelta(seconds=5),
         heartbeat_timeout: timedelta = timedelta(seconds=5),
         retry_policy: RetryPolicy = DEFAULT_RETRY_POLICY,
-        extra_headers: dict[str, str] | None = None,
+        request: dict[str, Any] | None = None,
     ) -> List[TaskMessage]:
         """
         Send a message to a task.
@@ -190,7 +190,7 @@ async def send_message(
             start_to_close_timeout: The start to close timeout for the message.
             heartbeat_timeout: The heartbeat timeout for the message.
             retry_policy: The retry policy for the message.
-            extra_headers: Additional HTTP headers to forward to the agent (filtered by agent's allowlist).
+            request: Additional request context including headers to forward to the agent.
 
         Returns:
             The message entry.
@@ -205,7 +205,7 @@ async def send_message(
                     content=content,
                     trace_id=trace_id,
                     parent_span_id=parent_span_id,
-                    extra_headers=extra_headers,
+                    request=request,
                 ),
                 response_type=TaskMessage,
                 start_to_close_timeout=start_to_close_timeout,
@@ -220,7 +220,7 @@ async def send_message(
                 content=content,
                 trace_id=trace_id,
                 parent_span_id=parent_span_id,
-                extra_headers=extra_headers,
+                request=request,
             )
 
     async def cancel_task(
@@ -234,7 +234,7 @@ async def cancel_task(
         start_to_close_timeout: timedelta = timedelta(seconds=5),
         heartbeat_timeout: timedelta = timedelta(seconds=5),
         retry_policy: RetryPolicy = DEFAULT_RETRY_POLICY,
-        extra_headers: dict[str, str] | None = None,
+        request: dict[str, Any] | None = None,
     ) -> Task:
         """
         Cancel a task by sending cancel request to the agent that owns the task.
@@ -249,7 +249,7 @@ async def cancel_task(
             start_to_close_timeout: The start to close timeout for the task.
             heartbeat_timeout: The heartbeat timeout for the task.
             retry_policy: The retry policy for the task.
-            extra_headers: Additional HTTP headers to forward to the agent (filtered by agent's allowlist).
+            request: Additional request context including headers to forward to the agent.
 
         Returns:
             The task entry.
@@ -268,7 +268,7 @@ async def cancel_task(
                     agent_name=agent_name,
                     trace_id=trace_id,
                     parent_span_id=parent_span_id,
-                    extra_headers=extra_headers,
+                    request=request,
                 ),
                 response_type=None,
                 start_to_close_timeout=start_to_close_timeout,
@@ -283,5 +283,5 @@ async def cancel_task(
                 agent_name=agent_name,
                 trace_id=trace_id,
                 parent_span_id=parent_span_id,
-                extra_headers=extra_headers,
+                request=request,
             )

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

@@ -26,69 +26,66 @@ def __init__(
 
     async def _get_agent_header_config(
         self, agent_name: str | None, agent_id: str | None
-    ) -> CustomHeadersConfig:
+    ) -> CustomHeadersConfig | None:
         """
         Get agent's header configuration from manifest.
         
-        For now, we'll return a default empty config since we need to implement
-        manifest loading. This provides secure-by-default behavior.
+        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
         """
-        # Default configuration - no headers allowed (secure by default)
+        # For now, return None to enable pass-through behavior
         # This will be replaced with actual manifest loading
-        return CustomHeadersConfig(
-            strategy="allowlist",
-            allowed_headers=[],
-            max_header_size=8192,
-            max_headers_count=50
-        )
+        return None
 
     def _filter_headers(
         self,
-        extra_headers: dict[str, str] | None,
-        config: CustomHeadersConfig
+        headers: dict[str, str] | None,
+        config: CustomHeadersConfig | None
     ) -> dict[str, str]:
         """
         Filter headers based on agent's configuration.
         
+        Pass-through by default: if no config provided, all headers are forwarded.
+        If config exists, only allowlisted headers are forwarded.
+        
         Args:
-            extra_headers: Headers to filter
-            config: Agent's header configuration
+            headers: Headers to filter
+            config: Agent's header configuration (None = pass-through all)
             
         Returns:
-            Filtered headers dictionary containing only allowed headers
+            Filtered headers dictionary
         """
-        if not extra_headers or not config.allowed_headers:
+        if not headers:
             return {}
+        
+        # Pass-through behavior: if no config, forward all headers
+        if config is None:
+            logger.debug("No header filtering config found, passing through all %d headers", len(headers))
+            return headers
 
-        filtered = {}
-        headers_added = 0
-
-        for header_name, header_value in extra_headers.items():
-            # Check if we've hit the max headers limit
-            if headers_added >= config.max_headers_count:
-                logger.warning("Reached maximum header count limit (%s), ignoring remaining headers", config.max_headers_count)
-                break
+        # Apply filtering based on allowlist
+        if not config.allowed_headers:
+            logger.debug("Empty allowlist in config, blocking all headers")
+            return {}
 
-            # Check against allowlist patterns
+        filtered = {}
+        for header_name, header_value in headers.items():
+            # 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
 
             if header_allowed:
-                # Apply size limits
-                if len(header_value) <= config.max_header_size:
-                    filtered[header_name] = header_value
-                    headers_added += 1
-                    logger.debug("Allowed header: %s", header_name)
-                else:
-                    logger.warning("Header '%s' exceeds size limit (%s > %s), ignoring", header_name, len(header_value), config.max_header_size)
+                filtered[header_name] = header_value
+                logger.debug("Allowed header: %s", header_name)
             else:
                 logger.debug("Header '%s' not in allowlist, ignoring", header_name)
 
+        logger.debug("Filtered %d headers from %d based on allowlist", len(filtered), len(headers))
         return filtered
 
     async def task_create(
@@ -99,7 +96,7 @@ async def task_create(
         params: dict[str, Any] | None = None,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
-        extra_headers: dict[str, str] | None = None,
+        request: dict[str, Any] | None = None,
     ) -> Task:
         trace = self._tracer.trace(trace_id=trace_id)
         async with trace.span(
@@ -114,9 +111,10 @@ async def task_create(
         ) as span:
             heartbeat_if_in_workflow("task create")
 
-            # Get agent's header configuration and filter headers
+            # 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(extra_headers, header_config)
+            filtered_headers = self._filter_headers(headers, header_config)
 
             if agent_name:
                 json_rpc_response = await self._agentex_client.agents.rpc_by_name(
@@ -155,7 +153,7 @@ async def message_send(
         task_name: str | None = None,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
-        extra_headers: dict[str, str] | None = None,
+        request: dict[str, Any] | None = None,
     ) -> List[TaskMessage]:
         trace = self._tracer.trace(trace_id=trace_id)
         async with trace.span(
@@ -171,9 +169,10 @@ async def message_send(
         ) as span:
             heartbeat_if_in_workflow("message send")
 
-            # Get agent's header configuration and filter headers
+            # 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(extra_headers, header_config)
+            filtered_headers = self._filter_headers(headers, header_config)
 
             if agent_name:
                 json_rpc_response = await self._agentex_client.agents.rpc_by_name(
@@ -222,7 +221,7 @@ async def event_send(
         task_name: str | None = None,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
-        extra_headers: dict[str, str] | None = None,
+        request: dict[str, Any] | None = None,
     ) -> Event:
         trace = self._tracer.trace(trace_id=trace_id)
         async with trace.span(
@@ -237,9 +236,10 @@ async def event_send(
         ) as span:
             heartbeat_if_in_workflow("event send")
 
-            # Get agent's header configuration and filter headers
+            # 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(extra_headers, header_config)
+            filtered_headers = self._filter_headers(headers, header_config)
 
             if agent_name:
                 json_rpc_response = await self._agentex_client.agents.rpc_by_name(
@@ -277,7 +277,7 @@ async def task_cancel(
         agent_name: str | None = None,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
-        extra_headers: dict[str, str] | None = None,
+        request: dict[str, Any] | None = None,
     ) -> Task:
         """
         Cancel a task by sending cancel request to the agent that owns the task.
@@ -318,9 +318,10 @@ async def task_cancel(
         ) as span:
             heartbeat_if_in_workflow("task cancel")
 
-            # Get agent's header configuration and filter headers
+            # 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(extra_headers, header_config)
+            filtered_headers = self._filter_headers(headers, header_config)
 
             # Build params for the agent (task identification)
             params = {}

src/agentex/lib/core/temporal/activities/adk/acp/acp_activities.py

@@ -26,31 +26,31 @@ class TaskCreateParams(BaseModelWithTraceParams):
     agent_id: str | None = None
     agent_name: str | None = None
     params: dict[str, Any] | None = None
-    extra_headers: dict[str, str] | None = None
+    request: dict[str, Any] | None = None
 
 
 class MessageSendParams(BaseModelWithTraceParams):
     agent_id: str | None = None
     agent_name: str | None = None
     task_id: str | None = None
     content: TaskMessageContent
-    extra_headers: dict[str, str] | None = None
+    request: dict[str, Any] | None = None
 
 
 class EventSendParams(BaseModelWithTraceParams):
     agent_id: str | None = None
     agent_name: str | None = None
     task_id: str | None = None
     content: TaskMessageContent
-    extra_headers: dict[str, str] | None = None
+    request: dict[str, Any] | None = None
 
 
 class TaskCancelParams(BaseModelWithTraceParams):
     task_id: str | None = None
     task_name: str | None = None
     agent_id: str | None = None
     agent_name: str | None = None
-    extra_headers: dict[str, str] | None = None
+    request: dict[str, Any] | None = None
 
 
 class ACPActivities:
@@ -66,7 +66,7 @@ async def task_create(self, params: TaskCreateParams) -> Task:
             params=params.params,
             trace_id=params.trace_id,
             parent_span_id=params.parent_span_id,
-            extra_headers=params.extra_headers,
+            request=params.request,
         )
 
     @activity.defn(name=ACPActivityName.MESSAGE_SEND)
@@ -78,7 +78,7 @@ async def message_send(self, params: MessageSendParams) -> List[TaskMessage]:
             content=params.content,
             trace_id=params.trace_id,
             parent_span_id=params.parent_span_id,
-            extra_headers=params.extra_headers,
+            request=params.request,
         )
 
     @activity.defn(name=ACPActivityName.EVENT_SEND)
@@ -90,7 +90,7 @@ async def event_send(self, params: EventSendParams) -> Event:
             content=params.content,
             trace_id=params.trace_id,
             parent_span_id=params.parent_span_id,
-            extra_headers=params.extra_headers,
+            request=params.request,
         )
 
     @activity.defn(name=ACPActivityName.TASK_CANCEL)
@@ -102,5 +102,5 @@ async def task_cancel(self, params: TaskCancelParams) -> Task:
             agent_name=params.agent_name,
             trace_id=params.trace_id,
             parent_span_id=params.parent_span_id,
-            extra_headers=params.extra_headers,
+            request=params.request,
         )