feat(agentkit): support agent-level pipeline_id

Author: digitallysavvy

README.md

@@ -96,6 +96,38 @@ def start_conversation() -> str:
 
 `Agora` generates the required ConvoAI REST auth and RTC join tokens automatically when you provide `app_id` and `app_certificate`. For supported Agora-managed models, leave vendor API keys unset; provide keys when you want BYOK.
 
+## AI Studio pipeline IDs
+
+Use `pipeline_id` when you want a published AI Studio pipeline to provide the base agent configuration:
+
+```python
+agent = Agent(
+    name="support",
+    pipeline_id="studio-pipeline-id",
+)
+
+session = agent.create_session(
+    client,
+    channel="support-room",
+    agent_uid="1",
+    remote_uids=["100"],
+)
+```
+
+You can override it per session:
+
+```python
+session = agent.create_session(
+    client,
+    channel="support-room",
+    agent_uid="1",
+    remote_uids=["100"],
+    pipeline_id="session-pipeline-id",
+)
+```
+
+AgentKit sends the resolved value as the top-level `/join` field `pipeline_id`, not inside `properties`. Explicit Agent config such as `with_llm()`, `with_tts()`, `with_stt()`, `with_mllm()`, and `advanced_features` may send `properties` fields that override the saved pipeline settings.
+
 ### BYOK version
 
 Use the same `Agent` builder shape, but provide credentials explicitly when you want vendor-managed billing and routing instead of Agora-managed models.

docs/reference/agent.md

@@ -27,12 +27,14 @@ Agent(
     labels: Optional[Dict[str, str]] = None,
     rtc: Optional[RtcConfig] = None,
     filler_words: Optional[FillerWordsConfig] = None,
+    pipeline_id: Optional[str] = None,
 )
 ```
 
 | Parameter | Type | Default | Description |
 |---|---|---|---|
 | `name` | `Optional[str]` | `None` | Agent name, used as default session name |
+| `pipeline_id` | `Optional[str]` | `None` | Published AI Studio pipeline ID used as this agent's base configuration |
 | `instructions` | `Optional[str]` | `None` | Deprecated. Use LLM vendor `system_messages` instead. |
 | `turn_detection` | `Optional[TurnDetectionConfig]` | `None` | Interaction language and turn detection configuration |
 | `interruption` | `Optional[InterruptionConfig]` | `None` | Unified interruption control configuration |
@@ -47,6 +49,8 @@ Agent(
 | `rtc` | `Optional[RtcConfig]` | `None` | RTC media encryption |
 | `filler_words` | `Optional[FillerWordsConfig]` | `None` | Filler words while waiting for LLM |
 
+`pipeline_id` is an AI Studio base configuration. Explicit Agent config such as `with_llm()`, `with_tts()`, `with_stt()`, `with_mllm()`, `advanced_features`, and other builder options may send fields in `properties` that override the saved pipeline settings. Session-level `pipeline_id` overrides the agent-level value.
+
 The Agent-level `instructions`, `greeting`, `failure_message`, `max_history`, and `greeting_configs` fields are compatibility shims. New code should configure those values on the LLM or MLLM vendor because that matches the core request schema.
 
 ## Builder Methods
@@ -202,6 +206,8 @@ create_session(
     token: Optional[str] = None,
     idle_timeout: Optional[int] = None,
     enable_string_uid: Optional[bool] = None,
+    preset: Optional[Union[str, Sequence[str]]] = None,
+    pipeline_id: Optional[str] = None,
     expires_in: Optional[int] = None,
 ) -> AgentSession
 ```
@@ -219,6 +225,10 @@ Creates an `AgentSession` bound to the given client and channel.
 | `expires_in` | `Optional[int]` | No | Token lifetime in seconds (default: `86400` = 24 h, Agora max). Only applies when the token is auto-generated. Use `expires_in_hours()` or `expires_in_minutes()` for clarity. Valid range: 1–86400. |
 | `idle_timeout` | `Optional[int]` | No | Idle timeout in seconds |
 | `enable_string_uid` | `Optional[bool]` | No | Enable string UIDs |
+| `preset` | `Optional[Union[str, Sequence[str]]]` | No | Advanced preset value for project-specific routing |
+| `pipeline_id` | `Optional[str]` | No | Published AI Studio pipeline ID for this session. Overrides `agent.pipeline_id`. |
+
+`pipeline_id` is sent as the top-level `/join` field `pipeline_id`, not inside `properties`.
 
 **Returns:** `AgentSession`
 

docs/reference/session.md

@@ -33,6 +33,11 @@ AgentSession(
     token: Optional[str] = None,
     idle_timeout: Optional[int] = None,
     enable_string_uid: Optional[bool] = None,
+    preset: Optional[Union[str, Sequence[str]]] = None,
+    pipeline_id: Optional[str] = None,
+    expires_in: Optional[int] = None,
+    debug: Optional[bool] = None,
+    warn: Optional[Callable[[str], None]] = None,
 )
 ```
 
@@ -51,6 +56,13 @@ AgentSession(
 | `token` | `Optional[str]` | No | Pre-built RTC token |
 | `idle_timeout` | `Optional[int]` | No | Idle timeout in seconds |
 | `enable_string_uid` | `Optional[bool]` | No | Enable string UIDs |
+| `preset` | `Optional[Union[str, Sequence[str]]]` | No | Advanced preset value for project-specific routing |
+| `pipeline_id` | `Optional[str]` | No | Published AI Studio pipeline ID for this session. Overrides `agent.pipeline_id`. |
+| `expires_in` | `Optional[int]` | No | Auto-generated token lifetime in seconds |
+| `debug` | `Optional[bool]` | No | Enable debug logging of the start request |
+| `warn` | `Optional[Callable[[str], None]]` | No | Custom warning sink |
+
+`pipeline_id` is sent as the top-level `/join` field `pipeline_id`, not inside `properties`. If unset, `AgentSession.start()` uses the agent-level value from `Agent(..., pipeline_id=...)`.
 
 ## Methods
 

src/agora_agent/agentkit/agent.py

@@ -343,8 +343,10 @@ def __init__(
         rtc: typing.Optional[RtcConfig] = None,
         filler_words: typing.Optional[FillerWordsConfig] = None,
         greeting_configs: typing.Optional[LlmGreetingConfigs] = None,
+        pipeline_id: typing.Optional[str] = None,
     ):
         self._name = name
+        self._pipeline_id = pipeline_id
         self._instructions = instructions
         self._greeting = greeting
         self._failure_message = failure_message
@@ -609,6 +611,11 @@ def _resolved_parameters(self) -> typing.Optional[typing.Union[SessionParams, Se
     def name(self) -> typing.Optional[str]:
         return self._name
 
+    @property
+    def pipeline_id(self) -> typing.Optional[str]:
+        """Published AI Studio pipeline ID used as this agent's base configuration."""
+        return self._pipeline_id
+
     @property
     def llm(self) -> typing.Optional[typing.Dict[str, typing.Any]]:
         return self._llm
@@ -693,6 +700,7 @@ def filler_words(self) -> typing.Optional[FillerWordsConfig]:
     def config(self) -> typing.Dict[str, typing.Any]:
         return {
             "name": self._name,
+            "pipeline_id": self._pipeline_id,
             "instructions": self._instructions,
             "greeting": self._greeting,
             "failure_message": self._failure_message,
@@ -945,6 +953,7 @@ def _resolve_turn_detection_config(self) -> TurnDetectionConfig:
     def _clone(self) -> "Agent":
         new_agent = Agent.__new__(Agent)
         new_agent._name = self._name
+        new_agent._pipeline_id = self._pipeline_id
         new_agent._llm = self._llm
         new_agent._tts = self._tts
         new_agent._stt = self._stt

src/agora_agent/agentkit/agent_session.py

@@ -52,7 +52,8 @@ class AgentSessionOptions(_AgentSessionRequiredOptions, total=False):
 
     Optional fields
     ---------------
-    app_certificate, token, idle_timeout, enable_string_uid, expires_in
+    app_certificate, token, idle_timeout, enable_string_uid, preset,
+    pipeline_id, expires_in, debug, warn
     """
 
     app_certificate: str
@@ -290,14 +291,18 @@ def _is_mllm_mode(self) -> bool:
             return True
         return mllm is not None
 
-    def _build_start_properties(self, token_opts: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:
+    def _build_start_properties(
+        self,
+        token_opts: typing.Dict[str, typing.Any],
+        skip_vendor_validation: bool,
+    ) -> typing.Dict[str, typing.Any]:
         base_properties = self._agent.to_properties(
             channel=self._channel,
             agent_uid=self._agent_uid,
             remote_uids=self._remote_uids,
             idle_timeout=self._idle_timeout,
             enable_string_uid=self._enable_string_uid,
-            skip_vendor_validation=True,
+            skip_vendor_validation=skip_vendor_validation,
             **token_opts,
         )
         properties = self._dump_model(base_properties)
@@ -445,6 +450,7 @@ def start(self) -> str:
         self._status = "starting"
 
         try:
+            pipeline_id = self._pipeline_id if self._pipeline_id is not None else self._agent.pipeline_id
             if self._token:
                 token_opts: typing.Dict[str, typing.Any] = {"token": self._token}
             else:
@@ -454,7 +460,7 @@ def start(self) -> str:
                     "expires_in": self._expires_in,
                 }
 
-            properties = self._build_start_properties(token_opts)
+            properties = self._build_start_properties(token_opts, skip_vendor_validation=bool(self._preset or pipeline_id))
             resolved_preset, resolved_properties = resolve_session_presets(
                 self._preset,
                 properties,
@@ -466,7 +472,7 @@ def start(self) -> str:
                     "appid": self._app_id,
                     "name": self._name,
                     "preset": resolved_preset,
-                    "pipeline_id": self._pipeline_id,
+                    "pipeline_id": pipeline_id,
                     "properties": resolved_properties,
                 })
 
@@ -480,7 +486,7 @@ def start(self) -> str:
                 name=self._name,
                 properties=request_properties,
                 preset=resolved_preset,
-                pipeline_id=self._pipeline_id,
+                pipeline_id=pipeline_id,
                 request_options=self._request_options(),
             )
 
@@ -766,6 +772,7 @@ async def start(self) -> str:
         self._status = "starting"
 
         try:
+            pipeline_id = self._pipeline_id if self._pipeline_id is not None else self._agent.pipeline_id
             if self._token:
                 token_opts: typing.Dict[str, typing.Any] = {"token": self._token}
             else:
@@ -775,7 +782,7 @@ async def start(self) -> str:
                     "expires_in": self._expires_in,
                 }
 
-            properties = self._build_start_properties(token_opts)
+            properties = self._build_start_properties(token_opts, skip_vendor_validation=bool(self._preset or pipeline_id))
             resolved_preset, resolved_properties = resolve_session_presets(
                 self._preset,
                 properties,
@@ -787,7 +794,7 @@ async def start(self) -> str:
                     "appid": self._app_id,
                     "name": self._name,
                     "preset": resolved_preset,
-                    "pipeline_id": self._pipeline_id,
+                    "pipeline_id": pipeline_id,
                     "properties": resolved_properties,
                 })
 
@@ -801,7 +808,7 @@ async def start(self) -> str:
                 name=self._name,
                 properties=request_properties,
                 preset=resolved_preset,
-                pipeline_id=self._pipeline_id,
+                pipeline_id=pipeline_id,
                 request_options=self._request_options(),
             )
 

tests/custom/test_pipeline_id.py

@@ -0,0 +1,123 @@
+import pytest
+
+from agora_agent import Agent
+
+
+def dump(value):
+    if hasattr(value, "model_dump"):
+        return value.model_dump(exclude_none=True)
+    if hasattr(value, "dict"):
+        return value.dict(exclude_none=True)
+    return value
+
+
+class StartResponse:
+    agent_id = "agent-id"
+
+
+class FakeAgentsClient:
+    def __init__(self):
+        self.calls = []
+
+    def start(self, appid, **kwargs):
+        self.calls.append({"appid": appid, **kwargs})
+        return StartResponse()
+
+
+class FakeAsyncAgentsClient:
+    def __init__(self):
+        self.calls = []
+
+    async def start(self, appid, **kwargs):
+        self.calls.append({"appid": appid, **kwargs})
+        return StartResponse()
+
+
+class FakeClient:
+    app_id = "appid"
+    app_certificate = None
+
+    def __init__(self, agents):
+        self.agents = agents
+
+
+def start_agent(agent, **overrides):
+    agents = FakeAgentsClient()
+    client = FakeClient(agents)
+    options = {
+        "channel": "channel",
+        "token": "token",
+        "agent_uid": "1",
+        "remote_uids": ["100"],
+        **overrides,
+    }
+
+    agent_id = agent.create_session(client, **options).start()
+
+    assert agent_id == "agent-id"
+    assert len(agents.calls) == 1
+    return agents.calls[0]
+
+
+def test_agent_pipeline_id_sends_top_level_pipeline_id() -> None:
+    call = start_agent(Agent(name="support", pipeline_id="studio-pipeline-id"))
+
+    assert call["appid"] == "appid"
+    assert call["name"] == "support"
+    assert call["pipeline_id"] == "studio-pipeline-id"
+    properties = dump(call["properties"])
+    assert properties["channel"] == "channel"
+    assert properties["token"] == "token"
+    assert properties["agent_rtc_uid"] == "1"
+    assert properties["remote_rtc_uids"] == ["100"]
+
+
+def test_session_pipeline_id_overrides_agent_pipeline_id() -> None:
+    call = start_agent(
+        Agent(name="support", pipeline_id="agent-pipeline"),
+        pipeline_id="session-pipeline",
+    )
+
+    assert call["pipeline_id"] == "session-pipeline"
+
+
+def test_agent_pipeline_id_skips_missing_vendor_validation() -> None:
+    call = start_agent(Agent(name="support", pipeline_id="studio-pipeline-id"))
+
+    assert call["pipeline_id"] == "studio-pipeline-id"
+
+
+def test_pipeline_id_is_not_sent_inside_properties() -> None:
+    call = start_agent(Agent(name="support", pipeline_id="studio-pipeline-id"))
+
+    assert call["pipeline_id"] == "studio-pipeline-id"
+    assert "pipeline_id" not in dump(call["properties"])
+
+
+def test_pipeline_id_survives_builder_clone() -> None:
+    agent = Agent(name="support", pipeline_id="studio-pipeline-id").with_tools(True)
+
+    assert agent.pipeline_id == "studio-pipeline-id"
+    call = start_agent(agent)
+
+    assert call["pipeline_id"] == "studio-pipeline-id"
+    assert dump(call["properties"])["advanced_features"] == {"enable_tools": True}
+
+
+@pytest.mark.asyncio
+async def test_async_session_uses_agent_pipeline_id() -> None:
+    agents = FakeAsyncAgentsClient()
+    client = FakeClient(agents)
+    agent = Agent(name="support", pipeline_id="studio-pipeline-id")
+
+    agent_id = await agent.create_async_session(
+        client,
+        channel="channel",
+        token="token",
+        agent_uid="1",
+        remote_uids=["100"],
+    ).start()
+
+    assert agent_id == "agent-id"
+    assert agents.calls[0]["pipeline_id"] == "studio-pipeline-id"
+    assert "pipeline_id" not in dump(agents.calls[0]["properties"])