feat: add end user id field to rpc spec

agentex/openapi.yaml

@@ -4075,6 +4075,16 @@ components:
           title: Task Name
           description: The name of the task to cancel. Either this or task_id must
             be provided.
+        end_user_id:
+          anyOf:
+          - type: string
+            maxLength: 256
+          - type: 'null'
+          title: End User Id
+          description: Opaque identifier for the end user this call is made on behalf
+            of. Not persisted on the task row, so supply it on every call that should
+            be attributed. Forwarded to the agent inside the ACP payload, which stamps
+            it onto the agent's trace spans as __end_user_id__.
       type: object
       title: CancelTaskRequest
     CheckpointListItem:
@@ -4520,6 +4530,16 @@ components:
           description: Caller-provided metadata to persist on the task row. Only applied
             at task creation; ignored if a task with this name already exists. Forwarded
             to the agent inside the ACP payload for backward compatibility.
+        end_user_id:
+          anyOf:
+          - type: string
+            maxLength: 256
+          - type: 'null'
+          title: End User Id
+          description: Opaque identifier for the end user this call is made on behalf
+            of. Not persisted on the task row, so supply it on every call that should
+            be attributed. Forwarded to the agent inside the ACP payload, which stamps
+            it onto the agent's trace spans as __end_user_id__.
       type: object
       title: CreateTaskRequest
     DataContent:
@@ -5610,6 +5630,16 @@ components:
           - $ref: '#/components/schemas/TaskMessageContent'
           - type: 'null'
           description: The content to send to the event
+        end_user_id:
+          anyOf:
+          - type: string
+            maxLength: 256
+          - type: 'null'
+          title: End User Id
+          description: Opaque identifier for the end user this call is made on behalf
+            of. Not persisted on the task row, so supply it on every call that should
+            be attributed. Forwarded to the agent inside the ACP payload, which stamps
+            it onto the agent's trace spans as __end_user_id__.
       type: object
       title: SendEventRequest
     SendMessageRequest:
@@ -5641,6 +5671,16 @@ components:
           - type: 'null'
           title: Task Params
           description: The parameters for the task (only used when creating new tasks)
+        end_user_id:
+          anyOf:
+          - type: string
+            maxLength: 256
+          - type: 'null'
+          title: End User Id
+          description: Opaque identifier for the end user this call is made on behalf
+            of. Not persisted on the task row, so supply it on every call that should
+            be attributed. Forwarded to the agent inside the ACP payload, which stamps
+            it onto the agent's trace spans as __end_user_id__.
       type: object
       required:
       - content

agentex/src/api/schemas/agents_rpc.py

@@ -13,6 +13,17 @@
 
 logger = make_logger(__name__)
 
+# Bounded because the value is forwarded to the agent on every call and, for
+# Temporal agents, copied into activity headers and therefore into workflow
+# history repeatedly.
+END_USER_ID_MAX_LENGTH = 256
+END_USER_ID_DESCRIPTION = (
+    "Opaque identifier for the end user this call is made on behalf of. Not "
+    "persisted on the task row, so supply it on every call that should be "
+    "attributed. Forwarded to the agent inside the ACP payload, which stamps it "
+    "onto the agent's trace spans as __end_user_id__."
+)
+
 
 class AgentRPCMethod(str, Enum):
     EVENT_SEND = "event/send"
@@ -34,6 +45,11 @@ class CreateTaskRequest(BaseModel):
             "Forwarded to the agent inside the ACP payload for backward compatibility."
         ),
     )
+    end_user_id: str | None = Field(
+        None,
+        max_length=END_USER_ID_MAX_LENGTH,
+        description=END_USER_ID_DESCRIPTION,
+    )
 
 
 class CancelTaskRequest(BaseModel):
@@ -45,6 +61,11 @@ class CancelTaskRequest(BaseModel):
         None,
         description="The name of the task to cancel. Either this or task_id must be provided.",
     )
+    end_user_id: str | None = Field(
+        None,
+        max_length=END_USER_ID_MAX_LENGTH,
+        description=END_USER_ID_DESCRIPTION,
+    )
 
     @model_validator(mode="after")
     def validate_task_identifiers(self):
@@ -70,6 +91,11 @@ class SendMessageRequest(BaseModel):
         None,
         description="The parameters for the task (only used when creating new tasks)",
     )
+    end_user_id: str | None = Field(
+        None,
+        max_length=END_USER_ID_MAX_LENGTH,
+        description=END_USER_ID_DESCRIPTION,
+    )
 
     @model_validator(mode="after")
     def validate_task_identifiers(self):
@@ -88,6 +114,11 @@ class SendEventRequest(BaseModel):
     content: TaskMessageContent | None = Field(
         None, description="The content to send to the event"
     )
+    end_user_id: str | None = Field(
+        None,
+        max_length=END_USER_ID_MAX_LENGTH,
+        description=END_USER_ID_DESCRIPTION,
+    )
 
     @model_validator(mode="after")
     def validate_task_identifiers(self):

agentex/src/domain/entities/agents_rpc.py

@@ -4,6 +4,8 @@
 from pydantic import BaseModel, Field, model_validator
 
 from src.api.schemas.agents_rpc import (
+    END_USER_ID_DESCRIPTION,
+    END_USER_ID_MAX_LENGTH,
     AgentRPCRequest,
     CancelTaskRequest,
     CreateTaskRequest,
@@ -30,6 +32,12 @@ class AgentRPCMethod(str, Enum):
     EVENT_SEND = "event/send"
 
 
+ACP_END_USER_ID_DESCRIPTION = (
+    "The end user this call was made on behalf of, when the caller supplied one. "
+    "The agent SDK stamps it onto trace spans as __end_user_id__."
+)
+
+
 class CreateTaskParams(BaseModel):
     """Parameters for task/create method"""
 
@@ -42,6 +50,7 @@ class CreateTaskParams(BaseModel):
         None,
         description="The parameters for the task as inputted by the user",
     )
+    end_user_id: str | None = Field(None, description=ACP_END_USER_ID_DESCRIPTION)
 
 
 class SendMessageParams(BaseModel):
@@ -58,6 +67,7 @@ class SendMessageParams(BaseModel):
     stream: bool = Field(
         False, description="Whether to stream the message to the agent"
     )
+    end_user_id: str | None = Field(None, description=ACP_END_USER_ID_DESCRIPTION)
 
 
 class SendEventParams(BaseModel):
@@ -73,6 +83,7 @@ class SendEventParams(BaseModel):
         None,
         description="Additional request context including headers",
     )
+    end_user_id: str | None = Field(None, description=ACP_END_USER_ID_DESCRIPTION)
 
 
 class CancelTaskParams(BaseModel):
@@ -83,6 +94,7 @@ class CancelTaskParams(BaseModel):
         description="The agent that the task was sent to",
     )
     task: TaskEntity = Field(..., description="The task that was cancelled")
+    end_user_id: str | None = Field(None, description=ACP_END_USER_ID_DESCRIPTION)
 
 
 ACP_TYPE_TO_ALLOWED_RPC_METHODS = {
@@ -113,6 +125,11 @@ class CreateTaskRequestEntity(BaseModel):
             "Forwarded to the agent inside the ACP payload for backward compatibility."
         ),
     )
+    end_user_id: str | None = Field(
+        None,
+        max_length=END_USER_ID_MAX_LENGTH,
+        description=END_USER_ID_DESCRIPTION,
+    )
 
 
 class CancelTaskRequestEntity(BaseModel):
@@ -124,6 +141,11 @@ class CancelTaskRequestEntity(BaseModel):
         None,
         description="The name of the task to cancel. Either this or task_id must be provided.",
     )
+    end_user_id: str | None = Field(
+        None,
+        max_length=END_USER_ID_MAX_LENGTH,
+        description=END_USER_ID_DESCRIPTION,
+    )
 
     @model_validator(mode="after")
     def validate_task_identifiers(self):
@@ -149,6 +171,11 @@ class SendMessageRequestEntity(BaseModel):
         None,
         description="The parameters for the task (only used when creating new tasks)",
     )
+    end_user_id: str | None = Field(
+        None,
+        max_length=END_USER_ID_MAX_LENGTH,
+        description=END_USER_ID_DESCRIPTION,
+    )
 
     @model_validator(mode="after")
     def validate_task_identifiers(self):
@@ -167,6 +194,11 @@ class SendEventRequestEntity(BaseModel):
     content: TaskMessageContentEntity | None = Field(
         None, description="The content to send to the event"
     )
+    end_user_id: str | None = Field(
+        None,
+        max_length=END_USER_ID_MAX_LENGTH,
+        description=END_USER_ID_DESCRIPTION,
+    )
 
     @model_validator(mode="after")
     def validate_task_identifiers(self):
@@ -193,13 +225,15 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self:
                 name=request.params.root.name,
                 params=request.params.root.params,
                 task_metadata=request.params.root.task_metadata,
+                end_user_id=request.params.root.end_user_id,
             )
         elif request.method == AgentRPCMethod.TASK_CANCEL and isinstance(
             request.params.root, CancelTaskRequest
         ):
             params = CancelTaskRequestEntity(
                 task_id=request.params.root.task_id,
                 task_name=request.params.root.task_name,
+                end_user_id=request.params.root.end_user_id,
             )
         elif request.method == AgentRPCMethod.MESSAGE_SEND:
             content_entity = convert_task_message_content_to_entity(
@@ -211,6 +245,7 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self:
                 content=content_entity,
                 stream=request.params.root.stream,
                 task_params=request.params.root.task_params,
+                end_user_id=request.params.root.end_user_id,
             )
         elif request.method == AgentRPCMethod.EVENT_SEND:
             if request.params.root.content is not None:
@@ -223,6 +258,7 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self:
                 task_id=request.params.root.task_id,
                 task_name=request.params.root.task_name,
                 content=content_entity,
+                end_user_id=request.params.root.end_user_id,
             )
         else:
             logger.error(f"Invalid method from request: {request}")

agentex/src/domain/services/agent_acp_service.py

@@ -304,12 +304,14 @@ async def create_task(
         task: TaskEntity,
         acp_url: str,
         params: dict[str, Any] | None = None,
+        end_user_id: str | None = None,
     ) -> dict[str, Any]:
         """Create a new task"""
         params = CreateTaskParams(
             agent=agent,
             task=task,
             params=params,
+            end_user_id=end_user_id,
         )
         headers = await self.get_headers(agent)
         return await self._call_jsonrpc(
@@ -326,13 +328,15 @@ async def send_message(
         task: TaskEntity,
         content: TaskMessageContentEntity,
         acp_url: str,
+        end_user_id: str | None = None,
     ) -> TaskMessageContentEntity:
         """Send a message to a running task"""
         params = SendMessageParams(
             agent=agent,
             task=task,
             content=content,
             stream=False,
+            end_user_id=end_user_id,
         )
         headers = await self.get_headers(agent)
         lock_key = hash((agent.id, task.id))
@@ -357,13 +361,15 @@ async def send_message_stream(
         task: TaskEntity,
         content: TaskMessageContentEntity,
         acp_url: str,
+        end_user_id: str | None = None,
     ) -> AsyncIterator[TaskMessageUpdateEntity]:
         """Send a message to a running task and stream the response"""
         params = SendMessageParams(
             agent=agent,
             task=task,
             content=content,
             stream=True,
+            end_user_id=end_user_id,
         )
         headers = await self.get_headers(agent)
         lock_key = hash((agent.id, task.id))
@@ -398,10 +404,14 @@ async def send_message_stream(
                 yield self._parse_task_message_update(chunk)
 
     async def cancel_task(
-        self, agent: AgentEntity, task: TaskEntity, acp_url: str
+        self,
+        agent: AgentEntity,
+        task: TaskEntity,
+        acp_url: str,
+        end_user_id: str | None = None,
     ) -> dict[str, Any]:
         """Cancel a running task"""
-        params = CancelTaskParams(agent=agent, task=task)
+        params = CancelTaskParams(agent=agent, task=task, end_user_id=end_user_id)
         headers = await self.get_headers(agent)
         return await self._call_jsonrpc(
             url=acp_url,
@@ -418,6 +428,7 @@ async def send_event(
         task: TaskEntity,
         acp_url: str,
         request_headers: dict[str, str] | None = None,
+        end_user_id: str | None = None,
     ) -> dict[str, Any]:
         """Send an event to a running task"""
 
@@ -426,6 +437,7 @@ async def send_event(
             task=task,
             event=event,
             request=None,
+            end_user_id=end_user_id,
         )
 
         headers = await self.get_headers(agent, request_headers)