livekit_api/services/
threads.rs

1// Copyright 2026 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::time::Duration;
16
17use livekit_protocol::agent::{agent_task, agent_thread};
18use livekit_token::get_env_keys;
19
20use super::{ServiceBase, ServiceResult};
21use crate::public_api::{self as api, call, next_cursor, opt, page_size, types, PROJECT};
22
23/// Agent threads of the project: conversations whose chat items, tasks and
24/// database outlive any one session.
25#[derive(Debug)]
26pub struct ThreadClient {
27    client: api::Client,
28}
29
30impl ThreadClient {
31    /// Authenticates with an API key and secret, signing a short-lived token per request.
32    pub fn with_api_key(host: &str, api_key: &str, api_secret: &str) -> Self {
33        Self::build(
34            host,
35            ServiceBase::with_api_key(api_key, api_secret),
36            crate::http_client::client(),
37        )
38    }
39
40    /// Authenticates with a pre-signed token, sent verbatim on every request.
41    pub fn with_token(host: &str, token: &str) -> Self {
42        Self::build(host, ServiceBase::with_token(token), crate::http_client::client())
43    }
44
45    /// Reads the API key and secret from the `LIVEKIT_API_KEY` and
46    /// `LIVEKIT_API_SECRET` environment variables.
47    pub fn new(host: &str) -> ServiceResult<Self> {
48        let (api_key, api_secret) = get_env_keys()?;
49        Ok(Self::with_api_key(host, &api_key, &api_secret))
50    }
51
52    /// Builds the client from an already-constructed HTTP client so the unified
53    /// [`LiveKitApi`](super::LiveKitApi) can share one connection pool across services.
54    pub(crate) fn build(host: &str, base: ServiceBase, http: crate::http_client::Client) -> Self {
55        Self { client: api::client(host, base, http) }
56    }
57
58    /// Overrides the default per-request timeout (10s) for calls on this client.
59    pub fn with_request_timeout(mut self, timeout: Duration) -> Self {
60        self.client.inner.timeout = timeout;
61        self
62    }
63
64    pub async fn create(
65        &self,
66        req: agent_thread::CreateRequest,
67    ) -> ServiceResult<agent_thread::Thread> {
68        let body = types::ThreadServiceCreateThreadBody {
69            attributes: req.attributes,
70            idle_ttl: req.idle_ttl,
71            scope: opt(&req.scope).map(str::to_owned),
72        };
73        let res = call(self.client.thread_service_create_thread(PROJECT, &body)).await?;
74        Ok(res.thread.unwrap_or_default())
75    }
76
77    pub async fn get(&self, thread_id: &str) -> ServiceResult<agent_thread::Thread> {
78        let res = call(self.client.thread_service_get_thread(PROJECT, thread_id)).await?;
79        Ok(res.thread.unwrap_or_default())
80    }
81
82    /// The threads of one scope (every scope when empty), newest first.
83    pub async fn list(
84        &self,
85        req: agent_thread::ListRequest,
86    ) -> ServiceResult<agent_thread::ListResponse> {
87        let res = call(self.client.thread_service_list_threads(
88            PROJECT,
89            opt(&req.page_token),
90            page_size(req.limit),
91            opt(&req.scope),
92        ))
93        .await?;
94        Ok(agent_thread::ListResponse {
95            threads: res.items,
96            next_page_token: next_cursor(res.page_info),
97        })
98    }
99
100    pub async fn update(
101        &self,
102        req: agent_thread::UpdateRequest,
103    ) -> ServiceResult<agent_thread::Thread> {
104        let body = types::ThreadServiceUpdateThreadBody {
105            attributes: req.attributes,
106            idle_ttl: req.idle_ttl,
107        };
108        let res =
109            call(self.client.thread_service_update_thread(PROJECT, &req.thread_id, &body)).await?;
110        Ok(res.thread.unwrap_or_default())
111    }
112
113    /// Deletes the thread and its database.
114    pub async fn delete(&self, thread_id: &str) -> ServiceResult<()> {
115        call(self.client.thread_service_delete_thread(PROJECT, thread_id)).await?;
116        Ok(())
117    }
118
119    /// The thread's chat items, oldest first unless `sort_order` says otherwise.
120    pub async fn items(
121        &self,
122        req: agent_thread::GetItemsRequest,
123    ) -> ServiceResult<agent_thread::GetItemsResponse> {
124        let sort_order = match agent_thread::SortOrder::try_from(req.sort_order) {
125            Ok(agent_thread::SortOrder::Asc) => {
126                Some(types::LivekitPublicapiCommonV1SortOrder::SortOrderAsc)
127            }
128            Ok(agent_thread::SortOrder::Desc) => {
129                Some(types::LivekitPublicapiCommonV1SortOrder::SortOrderDesc)
130            }
131            _ => None,
132        };
133        let res = call(self.client.thread_service_get_thread_items(
134            PROJECT,
135            &req.thread_id,
136            opt(&req.agent),
137            opt(&req.page_token),
138            page_size(req.limit),
139            sort_order,
140        ))
141        .await?;
142        Ok(agent_thread::GetItemsResponse {
143            items: res.items,
144            next_page_token: next_cursor(res.page_info),
145        })
146    }
147
148    /// The thread's tasks, oldest first.
149    pub async fn tasks(
150        &self,
151        req: agent_task::GetTasksRequest,
152    ) -> ServiceResult<agent_task::GetTasksResponse> {
153        let status = req.status.and_then(|s| agent_task::Status::try_from(s).ok());
154        let res = call(self.client.thread_service_get_thread_tasks(
155            PROJECT,
156            &req.thread_id,
157            opt(&req.page_token),
158            page_size(req.limit),
159            status.as_ref(),
160        ))
161        .await?;
162        Ok(agent_task::GetTasksResponse {
163            tasks: res.items,
164            next_page_token: next_cursor(res.page_info),
165        })
166    }
167}
livekit_api/services/threads.rs

livekit_api/services/
threads.rs
