videosdk/resources/
agents.rs

1//! Dispatches deployed agents into rooms, and manages cloud deployments.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5
6use futures_util::Stream;
7use reqwest::Method;
8use serde::{Deserialize, Serialize};
9use serde_json::{json, Map, Value};
10
11use crate::client::{CallOptions, Client};
12use crate::common::{string_enum, MessageResponse};
13use crate::error::{Error, Result};
14use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
15use crate::query::QueryBuilder;
16use crate::resources::rooms::RoomWebhook;
17use crate::resources::{escape, random_uuid, slugify_name};
18
19const DEPLOYMENTS_PATH: &str = "/ai/v1/cloud/deployments";
20const VERSIONS_PATH: &str = "/ai/v1/cloud/versions";
21const SECRETS_PATH: &str = "/ai/v1/cloud/secrets";
22
23/* -------------------------------- dispatch ------------------------------------ */
24
25/// Per-dispatch room behavior.
26#[derive(Debug, Clone, Default, Serialize)]
27#[serde(rename_all = "camelCase")]
28pub struct AgentRoomOptions {
29    /// Whether to end the session when the agent leaves.
30    #[serde(skip_serializing_if = "Option::is_none")]
31    pub auto_end_session: Option<bool>,
32    /// How long the session may run, in seconds.
33    #[serde(skip_serializing_if = "Option::is_none")]
34    pub session_timeout_seconds: Option<u32>,
35    /// Whether the agent runs in playground mode.
36    #[serde(skip_serializing_if = "Option::is_none")]
37    pub playground: Option<bool>,
38    /// Whether the agent receives video.
39    #[serde(skip_serializing_if = "Option::is_none")]
40    pub vision: Option<bool>,
41    /// Whether the agent joins the meeting.
42    #[serde(skip_serializing_if = "Option::is_none")]
43    pub join_meeting: Option<bool>,
44}
45
46/// The parameters for [`AgentsResource::dispatch`] and
47/// [`connect`](AgentsResource::connect).
48#[derive(Debug, Clone, Default)]
49pub struct DispatchAgentParams {
50    /// The deployed agent to dispatch. Required.
51    pub agent_id: Option<String>,
52    /// The room to join. One of `meeting_id` or `room_id` is required.
53    pub meeting_id: Option<String>,
54    /// An alias of `meeting_id`.
55    pub room_id: Option<String>,
56    /// Per-dispatch room behavior.
57    pub room_options: Option<AgentRoomOptions>,
58    /// When present, also places an outbound SIP call as part of the dispatch.
59    pub sip_options: Option<Map<String, Value>>,
60    /// Free-form metadata handed to the agent.
61    pub metadata: Option<Map<String, Value>>,
62    /// Pins a specific deployment version.
63    pub version_id: Option<String>,
64    /// Resolves a no-code published version by tag.
65    pub version_tag: Option<String>,
66    /// Forwarded to the registry when set.
67    pub wait: Option<Value>,
68}
69
70/// The parameters for [`AgentsResource::general_dispatch`], a managed or
71/// low-code dispatch. Unlike the others, this ignores `version_tag` and `wait`.
72#[derive(Debug, Clone, Default)]
73pub struct GeneralDispatchAgentParams {
74    /// The deployed agent to dispatch.
75    pub agent_id: Option<String>,
76    /// The room to join. One of `meeting_id` or `room_id` is required.
77    pub meeting_id: Option<String>,
78    /// An alias of `meeting_id`.
79    pub room_id: Option<String>,
80    /// Per-dispatch room behavior.
81    pub room_options: Option<AgentRoomOptions>,
82    /// When present, also places an outbound SIP call.
83    pub sip_options: Option<Map<String, Value>>,
84    /// Free-form metadata handed to the agent.
85    pub metadata: Option<Map<String, Value>>,
86    /// The deployment version. Required.
87    pub version_id: Option<String>,
88}
89
90impl From<GeneralDispatchAgentParams> for DispatchAgentParams {
91    fn from(params: GeneralDispatchAgentParams) -> Self {
92        Self {
93            agent_id: params.agent_id,
94            meeting_id: params.meeting_id,
95            room_id: params.room_id,
96            room_options: params.room_options,
97            sip_options: params.sip_options,
98            metadata: params.metadata,
99            version_id: params.version_id,
100            version_tag: None,
101            wait: None,
102        }
103    }
104}
105
106#[derive(Debug, Serialize)]
107#[serde(rename_all = "camelCase")]
108struct DispatchWire {
109    #[serde(skip_serializing_if = "Option::is_none")]
110    agent_id: Option<String>,
111    #[serde(skip_serializing_if = "Option::is_none")]
112    room_options: Option<AgentRoomOptions>,
113    #[serde(skip_serializing_if = "Option::is_none")]
114    sip_options: Option<Map<String, Value>>,
115    #[serde(skip_serializing_if = "Option::is_none")]
116    metadata: Option<Map<String, Value>>,
117    #[serde(skip_serializing_if = "Option::is_none")]
118    version_id: Option<String>,
119    #[serde(skip_serializing_if = "Option::is_none")]
120    version_tag: Option<String>,
121    #[serde(skip_serializing_if = "Option::is_none")]
122    wait: Option<Value>,
123    /// Always sent, even though the caller may have supplied only `room_id`.
124    meeting_id: String,
125    #[serde(skip_serializing_if = "Option::is_none")]
126    room_id: Option<String>,
127}
128
129/// The result of dispatching an agent.
130#[derive(Debug, Clone, Deserialize)]
131#[serde(rename_all = "camelCase")]
132pub struct DispatchResult {
133    /// Whether the dispatch was accepted.
134    pub success: Option<bool>,
135    /// The dispatch job id.
136    pub job_id: Option<String>,
137    /// The dispatch status.
138    pub status: Option<String>,
139    /// The worker that picked up the job.
140    pub worker_id: Option<String>,
141    /// The room the agent joined.
142    pub room_id: Option<String>,
143    /// The agent that was dispatched.
144    pub agent_id: Option<String>,
145    /// Cloud-deployment details.
146    pub cloud: Option<Value>,
147    /// Any fields the server returned that this SDK does not model yet.
148    #[serde(flatten)]
149    pub extra: Map<String, Value>,
150}
151
152/// The registry configuration a worker uses to initialize.
153#[derive(Debug, Clone, Deserialize)]
154#[serde(rename_all = "camelCase")]
155pub struct AgentInitConfig {
156    /// The agent registry URL.
157    pub registry_url: String,
158}
159
160/* ------------------------------- deployments ---------------------------------- */
161
162string_enum! {
163    /// The compute profile of a deployment version.
164    AgentComputeProfile {
165        /// A small CPU profile. The default.
166        CPU_SMALL => "cpu-small",
167        /// A medium CPU profile.
168        CPU_MEDIUM => "cpu-medium",
169        /// A large CPU profile.
170        CPU_LARGE => "cpu-large",
171    }
172}
173
174/// A cloud agent deployment.
175#[derive(Debug, Clone, Deserialize)]
176#[serde(rename_all = "camelCase")]
177pub struct AgentDeployment {
178    /// The deployment id.
179    pub id: String,
180    /// The agent this deployment belongs to.
181    pub agent_id: Option<String>,
182    /// The deployment's display name.
183    pub name: Option<String>,
184    /// The template the deployment was created from.
185    pub template: Option<String>,
186    /// The deployment's status.
187    pub status: Option<String>,
188    /// The attached env-secret id.
189    pub env_secret: Option<String>,
190    /// Any fields the server returned that this SDK does not model yet.
191    #[serde(flatten)]
192    pub extra: Map<String, Value>,
193}
194
195/// A deployment version.
196#[derive(Debug, Clone, Deserialize)]
197#[serde(rename_all = "camelCase")]
198pub struct AgentDeploymentVersion {
199    /// The version id.
200    pub id: String,
201    /// The deployment this version belongs to.
202    pub deployment_id: Option<String>,
203    /// The agent this version belongs to.
204    pub agent_id: Option<String>,
205    /// The version's display name.
206    pub name: Option<String>,
207    /// The container image.
208    pub image: Option<Value>,
209    /// The region the version runs in.
210    pub region: Option<String>,
211    /// The minimum replica count.
212    pub min_replica: Option<u32>,
213    /// The maximum replica count.
214    pub max_replica: Option<u32>,
215    /// The version's status.
216    pub status: Option<String>,
217    /// Whether this version is active.
218    pub active: Option<bool>,
219    /// The version tag.
220    pub version_tag: Option<String>,
221    /// Any fields the server returned that this SDK does not model yet.
222    #[serde(flatten)]
223    pub extra: Map<String, Value>,
224}
225
226/// A single console-log entry.
227#[derive(Debug, Clone, Deserialize)]
228pub struct AgentLogEntry {
229    /// The log line.
230    pub log: String,
231    /// When the line was emitted.
232    pub timestamp: Option<String>,
233    /// Structured metadata attached to the line.
234    pub metadata: Option<Map<String, Value>>,
235}
236
237/// A container image for a deployment version.
238#[derive(Debug, Clone, Default, Serialize)]
239#[serde(rename_all = "camelCase")]
240pub struct DeployImage {
241    /// The full image reference, e.g. `registry.videosdk.live/acme/agent:1.4`.
242    pub image_url: String,
243    /// The registry host. Parsed out of `image_url` when absent.
244    #[serde(rename = "imageCR", skip_serializing_if = "Option::is_none")]
245    pub image_cr: Option<String>,
246    /// One of `public` (the default), `private` — which needs an
247    /// `image_pull_secret` — or `managed`.
248    #[serde(skip_serializing_if = "Option::is_none")]
249    pub image_type: Option<String>,
250    /// The pull-secret id, for a private registry.
251    #[serde(skip_serializing_if = "Option::is_none")]
252    pub image_pull_secret: Option<String>,
253}
254
255/// A container image for a deployment: either a reference string, or a full
256/// [`DeployImage`].
257///
258/// ```
259/// # use videosdk::{AgentImage, DeployImage};
260/// let image: AgentImage = "registry.videosdk.live/acme/agent:1.4".into();
261/// let private: AgentImage = DeployImage {
262///     image_url: "registry.example.com/agent".into(),
263///     image_pull_secret: Some("secret-1".into()),
264///     ..Default::default()
265/// }
266/// .into();
267/// ```
268#[derive(Debug, Clone)]
269pub enum AgentImage {
270    /// An image reference, e.g. `registry.videosdk.live/acme/agent:1.4`.
271    Reference(String),
272    /// A full image object, for a private or managed registry.
273    Full(DeployImage),
274}
275
276impl From<&str> for AgentImage {
277    fn from(reference: &str) -> Self {
278        AgentImage::Reference(reference.to_string())
279    }
280}
281
282impl From<String> for AgentImage {
283    fn from(reference: String) -> Self {
284        AgentImage::Reference(reference)
285    }
286}
287
288impl From<DeployImage> for AgentImage {
289    fn from(image: DeployImage) -> Self {
290        AgentImage::Full(image)
291    }
292}
293
294/// The replica scaling of a deployment.
295#[derive(Debug, Clone, Default)]
296pub struct AgentScaling {
297    /// The minimum replica count.
298    pub min: Option<u32>,
299    /// The maximum replica count. Must be at most 50, and at least `min`. The
300    /// server rejects values outside that range rather than clamping them.
301    pub max: Option<u32>,
302}
303
304/// The parameters for [`AgentDeploymentResource::create`].
305#[derive(Debug, Clone, Default)]
306pub struct DeployAgentParams {
307    /// The deployment name. Required.
308    pub name: String,
309    /// The container image. Required in practice.
310    pub image: Option<AgentImage>,
311    /// Environment variables for the agent, stored as a secret attached to the
312    /// deployment.
313    pub env: HashMap<String, String>,
314    /// The replica scaling.
315    pub scaling: Option<AgentScaling>,
316    /// The compute profile. Defaults to `cpu-small`.
317    pub profile: Option<AgentComputeProfile>,
318    /// The region. Defaults to `us002`.
319    pub region: Option<String>,
320    /// Attaches to an existing agent. A new one is generated when absent.
321    pub agent_id: Option<String>,
322    /// The version tag.
323    pub version_tag: Option<String>,
324    /// The session webhook. Its `events` list is effectively required.
325    pub webhook: Option<RoomWebhook>,
326    /// Names the generated env secret. Derived from `name` when absent.
327    pub env_secret_name: Option<String>,
328}
329
330/// The result of [`AgentDeploymentResource::create`].
331#[derive(Debug, Clone)]
332pub struct DeployedAgent {
333    /// The deployment that was created.
334    pub deployment: AgentDeployment,
335    /// The version created, and auto-activated, for this deploy.
336    pub version: AgentDeploymentVersion,
337    /// The env-secret id, when `env` was provided.
338    pub secret_id: Option<String>,
339}
340
341/// Infers `image_cr` and `image_type` when the caller did not set them.
342fn normalize_deploy_image(image: Option<AgentImage>) -> DeployImage {
343    let mut image = match image {
344        None => DeployImage::default(),
345        Some(AgentImage::Reference(reference)) => DeployImage {
346            image_url: reference,
347            ..Default::default()
348        },
349        Some(AgentImage::Full(image)) => image,
350    };
351
352    if image.image_cr.is_none() {
353        let first_segment = image.image_url.split('/').next().unwrap_or("");
354        // A registry host carries a `.` or a `:`; anything else is a Docker Hub
355        // short reference.
356        image.image_cr = Some(
357            if !first_segment.is_empty() && first_segment.contains(['.', ':']) {
358                first_segment.to_string()
359            } else {
360                "docker.io".to_string()
361            },
362        );
363    }
364
365    if image.image_type.is_none() {
366        image.image_type = Some(
367            if image.image_pull_secret.is_some() {
368                "private"
369            } else {
370                "public"
371            }
372            .to_string(),
373        );
374    }
375    image
376}
377
378/// The query parameters for [`AgentDeploymentResource::list`].
379#[derive(Debug, Clone, Default)]
380pub struct ListDeploymentsParams {
381    /// The 1-based page number.
382    pub page: Option<u32>,
383    /// Items per page.
384    pub per_page: Option<u32>,
385    /// An opaque cursor from a previous page.
386    pub cursor: Option<String>,
387    /// Filters by agent id.
388    pub agent_id: Option<String>,
389    /// A regex match on the deployment name.
390    pub name: Option<String>,
391    /// Filters by deployment status.
392    pub status: Option<String>,
393    /// Filters by deployment id.
394    pub deployment_id: Option<String>,
395}
396
397/// The query parameters for [`AgentDeploymentResource::logs`].
398#[derive(Debug, Clone, Default)]
399pub struct ListDeploymentLogsParams {
400    /// The 1-based page number.
401    pub page: Option<u32>,
402    /// Items per page.
403    pub per_page: Option<u32>,
404    /// An opaque cursor from a previous page.
405    pub cursor: Option<String>,
406    /// Filters by agent id.
407    pub agent_id: Option<String>,
408    /// Filters by version id.
409    pub version_id: Option<String>,
410    /// Filters by session id.
411    pub session_id: Option<String>,
412    /// Filters by user id.
413    pub user_id: Option<String>,
414    /// Filters by room id.
415    pub room_id: Option<String>,
416    /// The start of the query window, as an RFC 3339 timestamp.
417    ///
418    /// The server parses this with `new Date(...)`, which yields `NaN` for a
419    /// stringified epoch, so a formatted timestamp is required.
420    pub start: Option<String>,
421    /// The end of the query window, as an RFC 3339 timestamp.
422    pub end: Option<String>,
423    /// The sort order: `1` ascending, `-1` descending.
424    pub sort: Option<String>,
425}
426
427macro_rules! pagination {
428    ($name:ident) => {
429        impl $name {
430            fn pagination(&self) -> ListParams {
431                ListParams {
432                    page: self.page,
433                    per_page: self.per_page,
434                    cursor: self.cursor.clone(),
435                }
436            }
437        }
438    };
439}
440
441pagination!(ListDeploymentsParams);
442pagination!(ListDeploymentLogsParams);
443
444/// Cloud agent deployments. Reached via [`AgentsResource::deployment`].
445#[derive(Debug, Clone, Copy)]
446pub struct AgentDeploymentResource<'a> {
447    client: &'a Client,
448}
449
450impl<'a> AgentDeploymentResource<'a> {
451    /// Lists cloud agent deployments, one page at a time.
452    pub async fn list(&self, params: ListDeploymentsParams) -> Result<Page<AgentDeployment>> {
453        paginate(
454            self.fetcher(&params),
455            &params.pagination(),
456            "deployments",
457            None,
458        )
459        .await
460    }
461
462    /// Lists deployments, transparently fetching every page.
463    pub fn list_stream(
464        &self,
465        params: ListDeploymentsParams,
466    ) -> impl Stream<Item = Result<AgentDeployment>> + Send {
467        auto_page(
468            self.fetcher(&params),
469            params.pagination(),
470            "deployments",
471            None,
472        )
473    }
474
475    /// Fetches a single deployment by id.
476    ///
477    /// There is no get-by-id route, so this filters the list.
478    pub async fn get(&self, deployment_id: &str) -> Result<AgentDeployment> {
479        let page = self
480            .list(ListDeploymentsParams {
481                deployment_id: Some(deployment_id.to_string()),
482                per_page: Some(1),
483                ..Default::default()
484            })
485            .await?;
486
487        page.data
488            .iter()
489            .find(|deployment| deployment.id == deployment_id)
490            .or_else(|| page.data.first())
491            .cloned()
492            .ok_or_else(|| Error::not_found(format!("agent deployment {deployment_id} not found")))
493    }
494
495    /// Fetches the latest version of a deployment.
496    pub async fn get_latest_version(&self, deployment_id: &str) -> Result<AgentDeploymentVersion> {
497        let path = format!("{DEPLOYMENTS_PATH}/{}/latest", escape(deployment_id));
498        self.client
499            .json(Method::GET, &path, CallOptions::new())
500            .await
501    }
502
503    /// Lists a deployment's console logs, one page at a time.
504    ///
505    /// This is a historical query — the last 24 hours by default — not a live
506    /// stream.
507    pub async fn logs(
508        &self,
509        deployment_id: &str,
510        params: ListDeploymentLogsParams,
511    ) -> Result<Page<AgentLogEntry>> {
512        let fetcher = self.logs_fetcher(deployment_id, &params);
513        paginate(fetcher, &params.pagination(), "logs", None).await
514    }
515
516    /// Lists a deployment's console logs, transparently fetching every page.
517    pub fn logs_stream(
518        &self,
519        deployment_id: &str,
520        params: ListDeploymentLogsParams,
521    ) -> impl Stream<Item = Result<AgentLogEntry>> + Send {
522        let fetcher = self.logs_fetcher(deployment_id, &params);
523        auto_page(fetcher, params.pagination(), "logs", None)
524    }
525
526    /// Soft-deletes a deployment.
527    pub async fn delete(&self, deployment_id: &str, force: bool) -> Result<MessageResponse> {
528        let path = format!("{DEPLOYMENTS_PATH}/{}/delete", escape(deployment_id));
529        let body = json!({ "force": force });
530        self.client
531            .json(Method::POST, &path, CallOptions::json(&body)?)
532            .await
533    }
534
535    /// Deploys a pre-built agent image.
536    ///
537    /// This makes **three calls**: a secret holding `env` (only when `env` is
538    /// non-empty), the deployment, then its auto-activated version.
539    ///
540    /// It is **not transactional**. If a later call fails, the earlier artifacts
541    /// already exist — clean up with [`delete`](AgentDeploymentResource::delete).
542    pub async fn create(&self, params: DeployAgentParams) -> Result<DeployedAgent> {
543        let image = normalize_deploy_image(params.image);
544
545        // 1. The env secret, when there are environment variables to store.
546        let mut secret_id = None;
547        if !params.env.is_empty() {
548            let name = params.env_secret_name.clone().unwrap_or_else(|| {
549                let suffix: String = random_uuid().chars().take(8).collect();
550                format!("{}-env-{suffix}", slugify_name(&params.name))
551            });
552            let body = json!({"name": name, "keys": params.env, "type": "NORMAL"});
553
554            #[derive(Deserialize)]
555            struct Secret {
556                id: String,
557            }
558            let secret: Secret = self
559                .client
560                .data(Method::POST, SECRETS_PATH, CallOptions::json(&body)?)
561                .await?;
562            secret_id = Some(secret.id);
563        }
564
565        // 2. The deployment. Send `agentId`/`envSecret` only when set: the server
566        // auto-generates an agent id via `agentId ?? generateAgentId()`, which an
567        // explicit empty string would defeat — and the schema then rejects it.
568        let mut deploy_body = Map::new();
569        deploy_body.insert("name".into(), json!(params.name));
570        if let Some(agent_id) = params.agent_id.as_deref().filter(|id| !id.is_empty()) {
571            deploy_body.insert("agentId".into(), json!(agent_id));
572        }
573        if let Some(secret_id) = &secret_id {
574            deploy_body.insert("envSecret".into(), json!(secret_id));
575        }
576        let deployment: AgentDeployment = self
577            .client
578            .json(
579                Method::POST,
580                DEPLOYMENTS_PATH,
581                CallOptions::json(&deploy_body)?,
582            )
583            .await?;
584
585        // 3. The version, which the server activates for us.
586        let scaling = params.scaling.unwrap_or_default();
587        let version_body = VersionWire {
588            agent_id: deployment.agent_id.as_deref(),
589            deployment_id: Some(&deployment.id),
590            image,
591            profile: params.profile.unwrap_or(AgentComputeProfile::CPU_SMALL),
592            min_replica: scaling.min,
593            max_replica: scaling.max,
594            region: params.region.as_deref(),
595            version_tag: params.version_tag.as_deref(),
596            webhook: params.webhook.as_ref(),
597        };
598        let version: AgentDeploymentVersion = self
599            .client
600            .json(
601                Method::POST,
602                VERSIONS_PATH,
603                CallOptions::json(&version_body)?,
604            )
605            .await?;
606
607        Ok(DeployedAgent {
608            deployment,
609            version,
610            secret_id,
611        })
612    }
613
614    /// Re-points a deployment at a version, or stops it.
615    ///
616    /// `activate: true` rolls onto `version_id`; `activate: false` stops the
617    /// running version. Not needed after
618    /// [`create`](AgentDeploymentResource::create), which already activates.
619    pub async fn set_version_active(
620        &self,
621        version_id: &str,
622        activate: bool,
623        force: bool,
624    ) -> Result<MessageResponse> {
625        let path = format!("{VERSIONS_PATH}/{}/activate", escape(version_id));
626        let body = json!({"activate": activate, "force": force});
627        self.client
628            .json(Method::PUT, &path, CallOptions::json(&body)?)
629            .await
630    }
631
632    fn fetcher(&self, params: &ListDeploymentsParams) -> PageFetcher {
633        let client = self.client.clone();
634        let params = params.clone();
635        Arc::new(move |page, per_page| {
636            let client = client.clone();
637            let params = params.clone();
638            Box::pin(async move {
639                let query = QueryBuilder::new()
640                    .opt("page", page)
641                    .opt("perPage", per_page)
642                    .opt_str("agentId", params.agent_id.as_deref())
643                    .opt_str("name", params.name.as_deref())
644                    .opt_str("status", params.status.as_deref())
645                    .opt_str("deploymentId", params.deployment_id.as_deref())
646                    .into_pairs();
647                client
648                    .json::<Value>(
649                        Method::GET,
650                        DEPLOYMENTS_PATH,
651                        CallOptions::new().query(query),
652                    )
653                    .await
654            })
655        })
656    }
657
658    fn logs_fetcher(&self, deployment_id: &str, params: &ListDeploymentLogsParams) -> PageFetcher {
659        let client = self.client.clone();
660        let params = params.clone();
661        let path = format!("{DEPLOYMENTS_PATH}/{}/console-logs", escape(deployment_id));
662        Arc::new(move |page, per_page| {
663            let client = client.clone();
664            let params = params.clone();
665            let path = path.clone();
666            Box::pin(async move {
667                let query = QueryBuilder::new()
668                    .opt("page", page)
669                    .opt("perPage", per_page)
670                    .opt_str("agentId", params.agent_id.as_deref())
671                    .opt_str("versionId", params.version_id.as_deref())
672                    .opt_str("sessionId", params.session_id.as_deref())
673                    .opt_str("userId", params.user_id.as_deref())
674                    .opt_str("roomId", params.room_id.as_deref())
675                    .opt_str("start", params.start.as_deref())
676                    .opt_str("end", params.end.as_deref())
677                    .opt_str("sort", params.sort.as_deref())
678                    .into_pairs();
679                client
680                    .json::<Value>(Method::GET, &path, CallOptions::new().query(query))
681                    .await
682            })
683        })
684    }
685}
686
687#[derive(Debug, Serialize)]
688#[serde(rename_all = "camelCase")]
689struct VersionWire<'a> {
690    #[serde(skip_serializing_if = "Option::is_none")]
691    agent_id: Option<&'a str>,
692    #[serde(skip_serializing_if = "Option::is_none")]
693    deployment_id: Option<&'a str>,
694    image: DeployImage,
695    profile: AgentComputeProfile,
696    #[serde(skip_serializing_if = "Option::is_none")]
697    min_replica: Option<u32>,
698    #[serde(skip_serializing_if = "Option::is_none")]
699    max_replica: Option<u32>,
700    #[serde(skip_serializing_if = "Option::is_none")]
701    region: Option<&'a str>,
702    #[serde(skip_serializing_if = "Option::is_none")]
703    version_tag: Option<&'a str>,
704    #[serde(skip_serializing_if = "Option::is_none")]
705    webhook: Option<&'a RoomWebhook>,
706}
707
708/* ---------------------------------- facade ------------------------------------ */
709
710/// Dispatches deployed agents into rooms. Reached via [`Client::agents`].
711#[derive(Debug, Clone, Copy)]
712pub struct AgentsResource<'a> {
713    client: &'a Client,
714}
715
716impl<'a> AgentsResource<'a> {
717    pub(crate) fn new(client: &'a Client) -> Self {
718        Self { client }
719    }
720
721    /// Cloud agent deployments.
722    pub fn deployment(&self) -> AgentDeploymentResource<'a> {
723        AgentDeploymentResource {
724            client: self.client,
725        }
726    }
727
728    /// Dispatches an agent into a room.
729    pub async fn dispatch(&self, params: DispatchAgentParams) -> Result<DispatchResult> {
730        self.send("/v2/agent/dispatch", params).await
731    }
732
733    /// An alias of [`dispatch`](AgentsResource::dispatch), on its own route.
734    pub async fn connect(&self, params: DispatchAgentParams) -> Result<DispatchResult> {
735        self.send("/v2/agent/connect", params).await
736    }
737
738    /// Dispatches a managed or low-code agent. Requires `version_id`.
739    pub async fn general_dispatch(
740        &self,
741        params: GeneralDispatchAgentParams,
742    ) -> Result<DispatchResult> {
743        self.send("/v2/agent/general/dispatch", params.into()).await
744    }
745
746    /// Fetches the agent registry URL a worker uses to initialize.
747    pub async fn init_config(&self) -> Result<AgentInitConfig> {
748        self.client
749            .data(Method::POST, "/v2/agent/init-config", CallOptions::new())
750            .await
751    }
752
753    async fn send(&self, path: &str, params: DispatchAgentParams) -> Result<DispatchResult> {
754        // The two ids alias each other; the server wants both populated.
755        let meeting_id = params
756            .meeting_id
757            .as_deref()
758            .or(params.room_id.as_deref())
759            .filter(|id| !id.is_empty())
760            .ok_or_else(|| Error::validation("agents.dispatch() requires meeting_id (or room_id)"))?
761            .to_string();
762        let room_id = params
763            .room_id
764            .filter(|id| !id.is_empty())
765            .unwrap_or_else(|| meeting_id.clone());
766
767        let body = DispatchWire {
768            agent_id: params.agent_id,
769            room_options: params.room_options,
770            sip_options: params.sip_options,
771            metadata: params.metadata,
772            version_id: params.version_id,
773            version_tag: params.version_tag,
774            wait: params.wait,
775            meeting_id,
776            room_id: Some(room_id),
777        };
778        self.client
779            .data(Method::POST, path, CallOptions::json(&body)?)
780            .await
781    }
782}
783
784#[cfg(test)]
785mod tests {
786    use super::*;
787
788    #[test]
789    fn a_registry_host_is_parsed_out_of_the_image_url() {
790        let image = normalize_deploy_image(Some("registry.videosdk.live/acme/agent:1.4".into()));
791        assert_eq!(image.image_cr.as_deref(), Some("registry.videosdk.live"));
792        assert_eq!(image.image_type.as_deref(), Some("public"));
793
794        // A host with a port also counts.
795        let image = normalize_deploy_image(Some("localhost:5000/agent".into()));
796        assert_eq!(image.image_cr.as_deref(), Some("localhost:5000"));
797    }
798
799    #[test]
800    fn a_short_reference_falls_back_to_docker_hub() {
801        for reference in ["acme/agent:1.4", "agent", "library/agent"] {
802            let image = normalize_deploy_image(Some(reference.into()));
803            assert_eq!(image.image_cr.as_deref(), Some("docker.io"), "{reference}");
804        }
805    }
806
807    #[test]
808    fn a_pull_secret_makes_the_image_private() {
809        let image = normalize_deploy_image(Some(
810            DeployImage {
811                image_url: "registry.example.com/agent".into(),
812                image_pull_secret: Some("secret-1".into()),
813                ..Default::default()
814            }
815            .into(),
816        ));
817        assert_eq!(image.image_type.as_deref(), Some("private"));
818        assert_eq!(image.image_url, "registry.example.com/agent");
819    }
820
821    #[test]
822    fn an_explicit_image_object_is_never_overwritten() {
823        let image = normalize_deploy_image(Some(
824            DeployImage {
825                image_url: "registry.example.com/agent".into(),
826                image_cr: Some("custom.registry".into()),
827                image_type: Some("managed".into()),
828                image_pull_secret: None,
829            }
830            .into(),
831        ));
832        assert_eq!(image.image_url, "registry.example.com/agent");
833        assert_eq!(image.image_cr.as_deref(), Some("custom.registry"));
834        assert_eq!(image.image_type.as_deref(), Some("managed"));
835    }
836}
videosdk/resources/agents.rs

videosdk/resources/
agents.rs
