|
3 | 3 | import logging |
4 | 4 | import os |
5 | 5 | from contextlib import AsyncExitStack |
| 6 | +import secrets |
| 7 | +import string |
6 | 8 | from typing import Any, Optional |
7 | 9 |
|
8 | 10 | from agent_framework import ( |
@@ -155,6 +157,33 @@ def get_chat_client(self, chat_client) -> AzureAIAgentClient: |
155 | 157 | extra={"agent_id": chat_client.agent_id}, |
156 | 158 | ) |
157 | 159 | return chat_client |
| 160 | + def generate_assistant_id(self, prefix: str = "asst_", length: int = 24) -> str: |
| 161 | + """ |
| 162 | + Generate a unique ID like 'asst_jRgR5t2U7o8nUPkNGv5HWOgV'. |
| 163 | +
|
| 164 | + - prefix: leading string (defaults to 'asst_') |
| 165 | + - length: number of random characters after the prefix |
| 166 | + """ |
| 167 | + # URL-safe characters similar to what OpenAI-style IDs use |
| 168 | + alphabet = string.ascii_letters + string.digits # a-zA-Z0-9 |
| 169 | + |
| 170 | + # cryptographically strong randomness |
| 171 | + random_part = "".join(secrets.choice(alphabet) for _ in range(length)) |
| 172 | + return f"{prefix}{random_part}" |
| 173 | + |
| 174 | + def get_agent_id(self, chat_client) -> str: |
| 175 | + """Return the underlying agent ID.""" |
| 176 | + if chat_client and chat_client.agent_id is not None: |
| 177 | + return chat_client.agent_id |
| 178 | + if ( |
| 179 | + self._agent |
| 180 | + and self._agent.chat_client |
| 181 | + and self._agent.chat_client.agent_id is not None |
| 182 | + ): |
| 183 | + return self._agent.chat_client.agent_id # type: ignore |
| 184 | + id = self.generate_assistant_id() |
| 185 | + self.logger.info("Generated new agent ID: %s", id) |
| 186 | + return id |
158 | 187 |
|
159 | 188 | async def get_database_team_agent(self) -> Optional[AzureAIAgentClient]: |
160 | 189 | """Retrieve existing team agent from database, if any.""" |
|
0 commit comments