[Feat] New API Provider - Add Azure AI Foundry Agents on /chat/completions, /responses, /messages + Agent Gateway (#17845)

* init get_azure_ai_route * init AzureAIAgentsConfig * init AzureAIAgentsConfig * AzureAIAgentsHandler * test_azure_ai_agents_acompletion_non_streaming * test_azure_ai_agents_acompletion_streaming * fix stream * _process_sse_stream * Azure AI Foundry Agents * init Azure AI Foundry Agent * fix code QA checks * fix api key * docs fix
2025-12-11 15:21:28 -08:00 · 2025-12-11 15:21:28 -08:00 · cca21c0926
commit cca21c0926
parent 8041e373d6
10 changed files with 1651 additions and 3 deletions
--- a/docs/my-website/docs/providers/azure_ai_agents.md
+++ b/docs/my-website/docs/providers/azure_ai_agents.md
@ -0,0 +1,292 @@
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+# Azure AI Foundry Agents
+
+Call Azure AI Foundry Agents in the OpenAI Request/Response format.
+
+| Property | Details |
+|----------|---------|
+| Description | Azure AI Foundry Agents provides hosted agent runtimes that can execute agentic workflows with foundation models, tools, and code interpreters. |
+| Provider Route on LiteLLM | `azure_ai/agents/{AGENT_ID}` |
+| Provider Doc | [Azure AI Foundry Agents ↗](https://learn.microsoft.com/en-us/rest/api/aifoundry/aiagents/create-thread-and-run/create-thread-and-run) |
+
+## Quick Start
+
+### Model Format to LiteLLM
+
+To call an Azure AI Foundry Agent through LiteLLM, use the following model format.
+
+Here the `model=azure_ai/agents/` tells LiteLLM to call the Azure AI Foundry Agent Service API.
+
+```shell showLineNumbers title="Model Format to LiteLLM"
+azure_ai/agents/{AGENT_ID}
+```
+
+**Example:**
+- `azure_ai/agents/asst_abc123`
+
+You can find the Agent ID in your Azure AI Foundry portal under Agents.
+
+### LiteLLM Python SDK
+
+```python showLineNumbers title="Basic Agent Completion"
+import litellm
+
+# Make a completion request to your Azure AI Foundry Agent
+response = litellm.completion(
+    model="azure_ai/agents/asst_abc123",
+    messages=[
+        {
+            "role": "user", 
+            "content": "Explain machine learning in simple terms"
+        }
+    ],
+    api_base="https://your-project.services.ai.azure.com",
+    api_key="your-api-key",
+)
+
+print(response.choices[0].message.content)
+print(f"Usage: {response.usage}")
+```
+
+```python showLineNumbers title="Streaming Agent Responses"
+import litellm
+
+# Stream responses from your Azure AI Foundry Agent
+response = await litellm.acompletion(
+    model="azure_ai/agents/asst_abc123",
+    messages=[
+        {
+            "role": "user",
+            "content": "What are the key principles of software architecture?"
+        }
+    ],
+    api_base="https://your-project.services.ai.azure.com",
+    api_key="your-api-key",
+    stream=True,
+)
+
+async for chunk in response:
+    if chunk.choices[0].delta.content:
+        print(chunk.choices[0].delta.content, end="")
+```
+
+### LiteLLM Proxy
+
+#### 1. Configure your model in config.yaml
+
+<Tabs>
+<TabItem value="config-yaml" label="config.yaml">
+
+```yaml showLineNumbers title="LiteLLM Proxy Configuration"
+model_list:
+  - model_name: azure-agent-1
+    litellm_params:
+      model: azure_ai/agents/asst_abc123
+      api_base: https://your-project.services.ai.azure.com
+      api_key: os.environ/AZURE_API_KEY
+
+  - model_name: azure-agent-math-tutor
+    litellm_params:
+      model: azure_ai/agents/asst_def456
+      api_base: https://your-project.services.ai.azure.com
+      api_key: os.environ/AZURE_API_KEY
+```
+
+</TabItem>
+</Tabs>
+
+#### 2. Start the LiteLLM Proxy
+
+```bash showLineNumbers title="Start LiteLLM Proxy"
+litellm --config config.yaml
+```
+
+#### 3. Make requests to your Azure AI Foundry Agents
+
+<Tabs>
+<TabItem value="curl" label="Curl">
+
+```bash showLineNumbers title="Basic Agent Request"
+curl http://localhost:4000/v1/chat/completions \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer $LITELLM_API_KEY" \
+  -d '{
+    "model": "azure-agent-1",
+    "messages": [
+      {
+        "role": "user", 
+        "content": "Summarize the main benefits of cloud computing"
+      }
+    ]
+  }'
+```
+
+```bash showLineNumbers title="Streaming Agent Request"
+curl http://localhost:4000/v1/chat/completions \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer $LITELLM_API_KEY" \
+  -d '{
+    "model": "azure-agent-math-tutor",
+    "messages": [
+      {
+        "role": "user",
+        "content": "What is 25 * 4?"
+      }
+    ],
+    "stream": true
+  }'
+```
+
+</TabItem>
+
+<TabItem value="openai-sdk" label="OpenAI Python SDK">
+
+```python showLineNumbers title="Using OpenAI SDK with LiteLLM Proxy"
+from openai import OpenAI
+
+# Initialize client with your LiteLLM proxy URL
+client = OpenAI(
+    base_url="http://localhost:4000",
+    api_key="your-litellm-api-key"
+)
+
+# Make a completion request to your Azure AI Foundry Agent
+response = client.chat.completions.create(
+    model="azure-agent-1",
+    messages=[
+      {
+        "role": "user",
+        "content": "What are best practices for API design?"
+      }
+    ]
+)
+
+print(response.choices[0].message.content)
+```
+
+```python showLineNumbers title="Streaming with OpenAI SDK"
+from openai import OpenAI
+
+client = OpenAI(
+    base_url="http://localhost:4000", 
+    api_key="your-litellm-api-key"
+)
+
+# Stream Agent responses
+stream = client.chat.completions.create(
+    model="azure-agent-math-tutor",
+    messages=[
+      {
+        "role": "user",
+        "content": "Explain the Pythagorean theorem"
+      }
+    ],
+    stream=True
+)
+
+for chunk in stream:
+    if chunk.choices[0].delta.content is not None:
+        print(chunk.choices[0].delta.content, end="")
+```
+
+</TabItem>
+</Tabs>
+
+## Environment Variables
+
+You can set the following environment variables to configure Azure AI Foundry Agents:
+
+| Variable | Description |
+|----------|-------------|
+| `AZURE_API_BASE` | The Azure AI Foundry project endpoint (e.g., `https://your-project.services.ai.azure.com`) |
+| `AZURE_API_KEY` | Your Azure AI Foundry API key |
+
+```bash
+export AZURE_API_BASE="https://your-project.services.ai.azure.com"
+export AZURE_API_KEY="your-api-key"
+```
+
+## Conversation Continuity (Thread Management)
+
+Azure AI Foundry Agents use threads to maintain conversation context. LiteLLM automatically manages threads for you, but you can also pass an existing thread ID to continue a conversation.
+
+```python showLineNumbers title="Continuing a Conversation"
+import litellm
+
+# First message creates a new thread
+response1 = await litellm.acompletion(
+    model="azure_ai/agents/asst_abc123",
+    messages=[{"role": "user", "content": "My name is Alice"}],
+    api_base="https://your-project.services.ai.azure.com",
+    api_key="your-api-key",
+)
+
+# Get the thread_id from the response
+thread_id = response1._hidden_params.get("thread_id")
+
+# Continue the conversation using the same thread
+response2 = await litellm.acompletion(
+    model="azure_ai/agents/asst_abc123",
+    messages=[{"role": "user", "content": "What's my name?"}],
+    api_base="https://your-project.services.ai.azure.com",
+    api_key="your-api-key",
+    thread_id=thread_id,  # Pass the thread_id to continue conversation
+)
+
+print(response2.choices[0].message.content)  # Should mention "Alice"
+```
+
+## Provider-specific Parameters
+
+Azure AI Foundry Agents support additional parameters that can be passed to customize the agent invocation.
+
+<Tabs>
+<TabItem value="sdk" label="SDK">
+
+```python showLineNumbers title="Using Agent-specific parameters"
+from litellm import completion
+
+response = litellm.completion(
+    model="azure_ai/agents/asst_abc123",
+    messages=[
+        {
+            "role": "user",
+            "content": "Analyze this data and provide insights",
+        }
+    ],
+    api_base="https://your-project.services.ai.azure.com",
+    api_key="your-api-key",
+    thread_id="thread_abc123",  # Optional: Continue existing conversation
+    instructions="Be concise and focus on key insights",  # Optional: Override agent instructions
+)
+```
+
+</TabItem>
+<TabItem value="proxy" label="Proxy">
+
+```yaml showLineNumbers title="LiteLLM Proxy Configuration with Parameters"
+model_list:
+  - model_name: azure-agent-analyst
+    litellm_params:
+      model: azure_ai/agents/asst_abc123
+      api_base: https://your-project.services.ai.azure.com
+      api_key: os.environ/AZURE_API_KEY
+      instructions: "Be concise and focus on key insights"
+```
+
+</TabItem>
+</Tabs>
+
+### Available Parameters
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `thread_id` | string | Optional thread ID to continue an existing conversation |
+| `instructions` | string | Optional instructions to override the agent's default instructions for this run |
+
+## Further Reading
+
+- [Azure AI Foundry Agents Documentation](https://learn.microsoft.com/en-us/azure/ai-services/agents/)
+- [Create Thread and Run API Reference](https://learn.microsoft.com/en-us/rest/api/aifoundry/aiagents/create-thread-and-run/create-thread-and-run)
--- a/docs/my-website/docs/providers/gemini.md
+++ b/docs/my-website/docs/providers/gemini.md
@ -1171,6 +1171,9 @@ When responding to Computer Use tool calls, include the URL and screenshot:
 }
 ```

+</TabItem>
+</Tabs>
+
 ### Environment Mapping

 | LiteLLM Input | Gemini API Value |
--- a/docs/my-website/sidebars.js
+++ b/docs/my-website/sidebars.js
@ -609,6 +609,7 @@ const sidebars = {
          label: "Azure AI",
          items: [
            "providers/azure_ai",
+            "providers/azure_ai_agents",
            "providers/azure_ocr",
            "providers/azure_document_intelligence",
            "providers/azure_ai_speech",
--- a/litellm/llms/azure_ai/agents/init.py
+++ b/litellm/llms/azure_ai/agents/init.py
@ -0,0 +1,11 @@
+from litellm.llms.azure_ai.agents.handler import azure_ai_agents_handler
+from litellm.llms.azure_ai.agents.transformation import (
+    AzureAIAgentsConfig,
+    AzureAIAgentsError,
+)
+
+__all__ = [
+    "AzureAIAgentsConfig",
+    "AzureAIAgentsError",
+    "azure_ai_agents_handler",
+]
--- a/litellm/llms/azure_ai/agents/handler.py
+++ b/litellm/llms/azure_ai/agents/handler.py
@ -0,0 +1,540 @@
+"""
+Handler for Azure AI Agent Service API.
+
+This handler executes the multi-step agent flow:
+1. Create thread (or use existing)
+2. Add messages to thread
+3. Create and poll a run
+4. Retrieve the assistant's response messages
+
+Model format: azure_ai/agents/<agent_id>
+
+Supports both polling-based and native streaming (SSE) modes.
+"""
+
+import asyncio
+import json
+import time
+import uuid
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncIterator,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+)
+
+import httpx
+
+from litellm._logging import verbose_logger
+from litellm.llms.azure_ai.agents.transformation import (
+    AzureAIAgentsConfig,
+    AzureAIAgentsError,
+)
+from litellm.types.utils import ModelResponse
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
+    from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
+
+    LiteLLMLoggingObj = _LiteLLMLoggingObj
+else:
+    LiteLLMLoggingObj = Any
+    HTTPHandler = Any
+    AsyncHTTPHandler = Any
+
+
+class AzureAIAgentsHandler:
+    """
+    Handler for Azure AI Agent Service.
+    
+    Executes the complete agent flow which requires multiple API calls.
+    """
+
+    def __init__(self):
+        self.config = AzureAIAgentsConfig()
+
+    # -------------------------------------------------------------------------
+    # URL Builders
+    # -------------------------------------------------------------------------
+    def _build_thread_url(self, api_base: str, api_version: str) -> str:
+        return f"{api_base}/openai/threads?api-version={api_version}"
+
+    def _build_messages_url(self, api_base: str, thread_id: str, api_version: str) -> str:
+        return f"{api_base}/openai/threads/{thread_id}/messages?api-version={api_version}"
+
+    def _build_runs_url(self, api_base: str, thread_id: str, api_version: str) -> str:
+        return f"{api_base}/openai/threads/{thread_id}/runs?api-version={api_version}"
+
+    def _build_run_status_url(self, api_base: str, thread_id: str, run_id: str, api_version: str) -> str:
+        return f"{api_base}/openai/threads/{thread_id}/runs/{run_id}?api-version={api_version}"
+
+    def _build_list_messages_url(self, api_base: str, thread_id: str, api_version: str) -> str:
+        return f"{api_base}/openai/threads/{thread_id}/messages?api-version={api_version}"
+
+    def _build_create_thread_and_run_url(self, api_base: str, api_version: str) -> str:
+        """URL for the create-thread-and-run endpoint (supports streaming)."""
+        return f"{api_base}/openai/threads/runs?api-version={api_version}"
+
+    # -------------------------------------------------------------------------
+    # Response Helpers
+    # -------------------------------------------------------------------------
+    def _extract_content_from_messages(self, messages_data: dict) -> str:
+        """Extract assistant content from the messages response."""
+        for msg in messages_data.get("data", []):
+            if msg.get("role") == "assistant":
+                for content_item in msg.get("content", []):
+                    if content_item.get("type") == "text":
+                        return content_item.get("text", {}).get("value", "")
+        return ""
+
+    def _build_model_response(
+        self,
+        model: str,
+        content: str,
+        model_response: ModelResponse,
+        thread_id: str,
+        messages: List[Dict[str, Any]],
+    ) -> ModelResponse:
+        """Build the ModelResponse from agent output."""
+        from litellm.types.utils import Choices, Message, Usage
+
+        model_response.choices = [
+            Choices(finish_reason="stop", index=0, message=Message(content=content, role="assistant"))
+        ]
+        model_response.model = model
+
+        # Store thread_id for conversation continuity
+        if not hasattr(model_response, "_hidden_params") or model_response._hidden_params is None:
+            model_response._hidden_params = {}
+        model_response._hidden_params["thread_id"] = thread_id
+
+        # Estimate token usage
+        try:
+            from litellm.utils import token_counter
+
+            prompt_tokens = token_counter(model="gpt-3.5-turbo", messages=messages)
+            completion_tokens = token_counter(model="gpt-3.5-turbo", text=content, count_response_tokens=True)
+            setattr(
+                model_response,
+                "usage",
+                Usage(
+                    prompt_tokens=prompt_tokens,
+                    completion_tokens=completion_tokens,
+                    total_tokens=prompt_tokens + completion_tokens,
+                ),
+            )
+        except Exception as e:
+            verbose_logger.warning(f"Failed to calculate token usage: {str(e)}")
+
+        return model_response
+
+    def _prepare_completion_params(
+        self,
+        model: str,
+        api_base: str,
+        api_key: str,
+        optional_params: dict,
+        headers: Optional[dict],
+    ) -> tuple:
+        """Prepare common parameters for completion."""
+        if headers is None:
+            headers = {}
+        headers["Content-Type"] = "application/json"
+        if api_key:
+            headers["api-key"] = api_key
+
+        api_version = optional_params.get("api_version", self.config.DEFAULT_API_VERSION)
+        agent_id = self.config._get_agent_id(model, optional_params)
+        thread_id = optional_params.get("thread_id")
+        api_base = api_base.rstrip("/")
+
+        verbose_logger.debug(f"Azure AI Agents completion - api_base: {api_base}, agent_id: {agent_id}")
+
+        return headers, api_version, agent_id, thread_id, api_base
+
+    def _check_response(self, response: httpx.Response, expected_codes: List[int], error_msg: str):
+        """Check response status and raise error if not expected."""
+        if response.status_code not in expected_codes:
+            raise AzureAIAgentsError(status_code=response.status_code, message=f"{error_msg}: {response.text}")
+
+    # -------------------------------------------------------------------------
+    # Sync Completion
+    # -------------------------------------------------------------------------
+    def completion(
+        self,
+        model: str,
+        messages: List[Dict[str, Any]],
+        api_base: str,
+        api_key: str,
+        model_response: ModelResponse,
+        logging_obj: LiteLLMLoggingObj,
+        optional_params: dict,
+        litellm_params: dict,
+        timeout: float,
+        client: Optional[HTTPHandler] = None,
+        headers: Optional[dict] = None,
+    ) -> ModelResponse:
+        """Execute synchronous completion using Azure Agent Service."""
+        from litellm.llms.custom_httpx.http_handler import _get_httpx_client
+
+        if client is None:
+            client = _get_httpx_client(params={"ssl_verify": litellm_params.get("ssl_verify", None)})
+
+        headers, api_version, agent_id, thread_id, api_base = self._prepare_completion_params(
+            model, api_base, api_key, optional_params, headers
+        )
+
+        def make_request(method: str, url: str, json_data: Optional[dict] = None) -> httpx.Response:
+            if method == "GET":
+                return client.get(url=url, headers=headers)
+            return client.post(url=url, headers=headers, data=json.dumps(json_data) if json_data else None)
+
+        # Execute the agent flow
+        thread_id, content = self._execute_agent_flow_sync(
+            make_request=make_request,
+            api_base=api_base,
+            api_version=api_version,
+            agent_id=agent_id,
+            thread_id=thread_id,
+            messages=messages,
+            optional_params=optional_params,
+        )
+
+        return self._build_model_response(model, content, model_response, thread_id, messages)
+
+    def _execute_agent_flow_sync(
+        self,
+        make_request: Callable,
+        api_base: str,
+        api_version: str,
+        agent_id: str,
+        thread_id: Optional[str],
+        messages: List[Dict[str, Any]],
+        optional_params: dict,
+    ) -> Tuple[str, str]:
+        """Execute the agent flow synchronously. Returns (thread_id, content)."""
+        
+        # Step 1: Create thread if not provided
+        if not thread_id:
+            verbose_logger.debug(f"Creating thread at: {self._build_thread_url(api_base, api_version)}")
+            response = make_request("POST", self._build_thread_url(api_base, api_version), {})
+            self._check_response(response, [200, 201], "Failed to create thread")
+            thread_id = response.json()["id"]
+            verbose_logger.debug(f"Created thread: {thread_id}")
+
+        # At this point thread_id is guaranteed to be a string
+        assert thread_id is not None
+
+        # Step 2: Add messages to thread
+        for msg in messages:
+            if msg.get("role") in ["user", "system"]:
+                url = self._build_messages_url(api_base, thread_id, api_version)
+                response = make_request("POST", url, {"role": "user", "content": msg.get("content", "")})
+                self._check_response(response, [200, 201], "Failed to add message")
+
+        # Step 3: Create run
+        run_payload = {"assistant_id": agent_id}
+        if "instructions" in optional_params:
+            run_payload["instructions"] = optional_params["instructions"]
+        
+        response = make_request("POST", self._build_runs_url(api_base, thread_id, api_version), run_payload)
+        self._check_response(response, [200, 201], "Failed to create run")
+        run_id = response.json()["id"]
+        verbose_logger.debug(f"Created run: {run_id}")
+
+        # Step 4: Poll for completion
+        status_url = self._build_run_status_url(api_base, thread_id, run_id, api_version)
+        for _ in range(self.config.MAX_POLL_ATTEMPTS):
+            response = make_request("GET", status_url)
+            self._check_response(response, [200], "Failed to get run status")
+            
+            status = response.json().get("status")
+            verbose_logger.debug(f"Run status: {status}")
+            
+            if status == "completed":
+                break
+            elif status in ["failed", "cancelled", "expired"]:
+                error_msg = response.json().get("last_error", {}).get("message", "Unknown error")
+                raise AzureAIAgentsError(status_code=500, message=f"Run {status}: {error_msg}")
+            
+            time.sleep(self.config.POLL_INTERVAL_SECONDS)
+        else:
+            raise AzureAIAgentsError(status_code=408, message="Run timed out waiting for completion")
+
+        # Step 5: Get messages
+        response = make_request("GET", self._build_list_messages_url(api_base, thread_id, api_version))
+        self._check_response(response, [200], "Failed to get messages")
+        
+        content = self._extract_content_from_messages(response.json())
+        return thread_id, content
+
+    # -------------------------------------------------------------------------
+    # Async Completion
+    # -------------------------------------------------------------------------
+    async def acompletion(
+        self,
+        model: str,
+        messages: List[Dict[str, Any]],
+        api_base: str,
+        api_key: str,
+        model_response: ModelResponse,
+        logging_obj: LiteLLMLoggingObj,
+        optional_params: dict,
+        litellm_params: dict,
+        timeout: float,
+        client: Optional[AsyncHTTPHandler] = None,
+        headers: Optional[dict] = None,
+    ) -> ModelResponse:
+        """Execute asynchronous completion using Azure Agent Service."""
+        import litellm
+        from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
+
+        if client is None:
+            client = get_async_httpx_client(
+                llm_provider=litellm.LlmProviders.AZURE_AI,
+                params={"ssl_verify": litellm_params.get("ssl_verify", None)},
+            )
+
+        headers, api_version, agent_id, thread_id, api_base = self._prepare_completion_params(
+            model, api_base, api_key, optional_params, headers
+        )
+
+        async def make_request(method: str, url: str, json_data: Optional[dict] = None) -> httpx.Response:
+            if method == "GET":
+                return await client.get(url=url, headers=headers)
+            return await client.post(url=url, headers=headers, data=json.dumps(json_data) if json_data else None)
+
+        # Execute the agent flow
+        thread_id, content = await self._execute_agent_flow_async(
+            make_request=make_request,
+            api_base=api_base,
+            api_version=api_version,
+            agent_id=agent_id,
+            thread_id=thread_id,
+            messages=messages,
+            optional_params=optional_params,
+        )
+
+        return self._build_model_response(model, content, model_response, thread_id, messages)
+
+    async def _execute_agent_flow_async(
+        self,
+        make_request: Callable,
+        api_base: str,
+        api_version: str,
+        agent_id: str,
+        thread_id: Optional[str],
+        messages: List[Dict[str, Any]],
+        optional_params: dict,
+    ) -> Tuple[str, str]:
+        """Execute the agent flow asynchronously. Returns (thread_id, content)."""
+        
+        # Step 1: Create thread if not provided
+        if not thread_id:
+            verbose_logger.debug(f"Creating thread at: {self._build_thread_url(api_base, api_version)}")
+            response = await make_request("POST", self._build_thread_url(api_base, api_version), {})
+            self._check_response(response, [200, 201], "Failed to create thread")
+            thread_id = response.json()["id"]
+            verbose_logger.debug(f"Created thread: {thread_id}")
+
+        # At this point thread_id is guaranteed to be a string
+        assert thread_id is not None
+
+        # Step 2: Add messages to thread
+        for msg in messages:
+            if msg.get("role") in ["user", "system"]:
+                url = self._build_messages_url(api_base, thread_id, api_version)
+                response = await make_request("POST", url, {"role": "user", "content": msg.get("content", "")})
+                self._check_response(response, [200, 201], "Failed to add message")
+
+        # Step 3: Create run
+        run_payload = {"assistant_id": agent_id}
+        if "instructions" in optional_params:
+            run_payload["instructions"] = optional_params["instructions"]
+        
+        response = await make_request("POST", self._build_runs_url(api_base, thread_id, api_version), run_payload)
+        self._check_response(response, [200, 201], "Failed to create run")
+        run_id = response.json()["id"]
+        verbose_logger.debug(f"Created run: {run_id}")
+
+        # Step 4: Poll for completion
+        status_url = self._build_run_status_url(api_base, thread_id, run_id, api_version)
+        for _ in range(self.config.MAX_POLL_ATTEMPTS):
+            response = await make_request("GET", status_url)
+            self._check_response(response, [200], "Failed to get run status")
+            
+            status = response.json().get("status")
+            verbose_logger.debug(f"Run status: {status}")
+            
+            if status == "completed":
+                break
+            elif status in ["failed", "cancelled", "expired"]:
+                error_msg = response.json().get("last_error", {}).get("message", "Unknown error")
+                raise AzureAIAgentsError(status_code=500, message=f"Run {status}: {error_msg}")
+            
+            await asyncio.sleep(self.config.POLL_INTERVAL_SECONDS)
+        else:
+            raise AzureAIAgentsError(status_code=408, message="Run timed out waiting for completion")
+
+        # Step 5: Get messages
+        response = await make_request("GET", self._build_list_messages_url(api_base, thread_id, api_version))
+        self._check_response(response, [200], "Failed to get messages")
+        
+        content = self._extract_content_from_messages(response.json())
+        return thread_id, content
+
+    # -------------------------------------------------------------------------
+    # Streaming Completion (Native SSE)
+    # -------------------------------------------------------------------------
+    async def acompletion_stream(
+        self,
+        model: str,
+        messages: List[Dict[str, Any]],
+        api_base: str,
+        api_key: str,
+        logging_obj: LiteLLMLoggingObj,
+        optional_params: dict,
+        litellm_params: dict,
+        timeout: float,
+        headers: Optional[dict] = None,
+    ) -> AsyncIterator:
+        """Execute async streaming completion using Azure Agent Service with native SSE."""
+        import litellm
+        from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
+
+        headers, api_version, agent_id, thread_id, api_base = self._prepare_completion_params(
+            model, api_base, api_key, optional_params, headers
+        )
+
+        # Build payload for create-thread-and-run with streaming
+        thread_messages = []
+        for msg in messages:
+            if msg.get("role") in ["user", "system"]:
+                thread_messages.append({
+                    "role": "user",
+                    "content": msg.get("content", "")
+                })
+
+        payload: Dict[str, Any] = {
+            "assistant_id": agent_id,
+            "stream": True,
+        }
+        
+        # Add thread with messages if we don't have an existing thread
+        if not thread_id:
+            payload["thread"] = {"messages": thread_messages}
+        
+        if "instructions" in optional_params:
+            payload["instructions"] = optional_params["instructions"]
+
+        url = self._build_create_thread_and_run_url(api_base, api_version)
+        verbose_logger.debug(f"Azure AI Agents streaming - URL: {url}")
+
+        # Use LiteLLM's async HTTP client for streaming
+        client = get_async_httpx_client(
+            llm_provider=litellm.LlmProviders.AZURE_AI,
+            params={"ssl_verify": litellm_params.get("ssl_verify", None)},
+        )
+
+        response = await client.post(
+            url=url,
+            headers=headers,
+            data=json.dumps(payload),
+            stream=True,
+        )
+
+        if response.status_code not in [200, 201]:
+            error_text = await response.aread()
+            raise AzureAIAgentsError(
+                status_code=response.status_code,
+                message=f"Streaming request failed: {error_text.decode()}"
+            )
+
+        async for chunk in self._process_sse_stream(response, model):
+            yield chunk
+
+    async def _process_sse_stream(
+        self,
+        response: httpx.Response,
+        model: str,
+    ) -> AsyncIterator:
+        """Process SSE stream and yield OpenAI-compatible streaming chunks."""
+        from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
+        
+        response_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"
+        created = int(time.time())
+        thread_id = None
+        
+        current_event = None
+        
+        async for line in response.aiter_lines():
+            line = line.strip()
+            
+            if line.startswith("event:"):
+                current_event = line[6:].strip()
+                continue
+            
+            if line.startswith("data:"):
+                data_str = line[5:].strip()
+                
+                if data_str == "[DONE]":
+                    # Send final chunk with finish_reason
+                    final_chunk = ModelResponseStream(
+                        id=response_id,
+                        created=created,
+                        model=model,
+                        object="chat.completion.chunk",
+                        choices=[
+                            StreamingChoices(
+                                finish_reason="stop",
+                                index=0,
+                                delta=Delta(content=None),
+                            )
+                        ],
+                    )
+                    if thread_id:
+                        final_chunk._hidden_params = {"thread_id": thread_id}
+                    yield final_chunk
+                    return
+                
+                try:
+                    data = json.loads(data_str)
+                except json.JSONDecodeError:
+                    continue
+                
+                # Extract thread_id from thread.created event
+                if current_event == "thread.created" and "id" in data:
+                    thread_id = data["id"]
+                    verbose_logger.debug(f"Stream created thread: {thread_id}")
+                
+                # Process message deltas - this is where the actual content comes
+                if current_event == "thread.message.delta":
+                    delta_content = data.get("delta", {}).get("content", [])
+                    for content_item in delta_content:
+                        if content_item.get("type") == "text":
+                            text_value = content_item.get("text", {}).get("value", "")
+                            if text_value:
+                                chunk = ModelResponseStream(
+                                    id=response_id,
+                                    created=created,
+                                    model=model,
+                                    object="chat.completion.chunk",
+                                    choices=[
+                                        StreamingChoices(
+                                            finish_reason=None,
+                                            index=0,
+                                            delta=Delta(content=text_value, role="assistant"),
+                                        )
+                                    ],
+                                )
+                                if thread_id:
+                                    chunk._hidden_params = {"thread_id": thread_id}
+                                yield chunk
+
+
+# Singleton instance
+azure_ai_agents_handler = AzureAIAgentsHandler()
--- a/litellm/llms/azure_ai/agents/transformation.py
+++ b/litellm/llms/azure_ai/agents/transformation.py
@ -0,0 +1,362 @@
+"""
+Transformation for Azure AI Agent Service API.
+
+Azure AI Agent Service provides an Assistants-like API for running agents.
+This follows the OpenAI Assistants pattern: create thread -> add messages -> create/poll run.
+
+Model format: azure_ai/agents/<agent_id>
+
+The API uses these endpoints:
+- POST /openai/threads - Create a thread
+- POST /openai/threads/{thread_id}/messages - Add message to thread
+- POST /openai/threads/{thread_id}/runs - Create a run
+- GET /openai/threads/{thread_id}/runs/{run_id} - Poll run status
+- GET /openai/threads/{thread_id}/messages - List messages in thread
+"""
+
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
+
+import httpx
+
+from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.prompt_templates.common_utils import (
+    convert_content_list_to_str,
+)
+from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
+from litellm.types.llms.openai import AllMessageValues
+from litellm.types.utils import ModelResponse
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
+    from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
+
+    LiteLLMLoggingObj = _LiteLLMLoggingObj
+else:
+    LiteLLMLoggingObj = Any
+    HTTPHandler = Any
+    AsyncHTTPHandler = Any
+
+
+class AzureAIAgentsError(BaseLLMException):
+    """Exception class for Azure AI Agent Service API errors."""
+
+    pass
+
+
+class AzureAIAgentsConfig(BaseConfig):
+    """
+    Configuration for Azure AI Agent Service API.
+
+    Azure AI Agent Service is a fully managed service for building AI agents
+    that can understand natural language and perform tasks.
+    
+    Model format: azure_ai/agents/<agent_id>
+    
+    The flow is:
+    1. Create a thread
+    2. Add user messages to the thread
+    3. Create and poll a run
+    4. Retrieve the assistant's response messages
+    """
+
+    # Default API version for Azure AI Agent Service
+    DEFAULT_API_VERSION = "2024-07-01-preview"
+    
+    # Polling configuration
+    MAX_POLL_ATTEMPTS = 60
+    POLL_INTERVAL_SECONDS = 1.0
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    @staticmethod
+    def is_azure_ai_agents_route(model: str) -> bool:
+        """
+        Check if the model is an Azure AI Agents route.
+        
+        Model format: azure_ai/agents/<agent_id>
+        """
+        return "agents/" in model
+
+    @staticmethod
+    def get_agent_id_from_model(model: str) -> str:
+        """
+        Extract agent ID from the model string.
+        
+        Model format: azure_ai/agents/<agent_id> -> <agent_id>
+        or: agents/<agent_id> -> <agent_id>
+        """
+        if "agents/" in model:
+            # Split on "agents/" and take the part after it
+            parts = model.split("agents/", 1)
+            if len(parts) == 2:
+                return parts[1]
+        return model
+
+    def _get_openai_compatible_provider_info(
+        self,
+        api_base: Optional[str],
+        api_key: Optional[str],
+    ) -> Tuple[Optional[str], Optional[str]]:
+        """
+        Get Azure AI Agent Service API base and key from params or environment.
+
+        Returns:
+            Tuple of (api_base, api_key)
+        """
+        from litellm.secret_managers.main import get_secret_str
+
+        api_base = api_base or get_secret_str("AZURE_AI_API_BASE")
+        api_key = api_key or get_secret_str("AZURE_AI_API_KEY")
+
+        return api_base, api_key
+
+    def get_supported_openai_params(self, model: str) -> List[str]:
+        """
+        Azure Agents supports minimal OpenAI params since it's an agent runtime.
+        """
+        return ["stream"]
+
+    def map_openai_params(
+        self,
+        non_default_params: dict,
+        optional_params: dict,
+        model: str,
+        drop_params: bool,
+    ) -> dict:
+        """
+        Map OpenAI params to Azure Agents params.
+        """
+        return optional_params
+
+    def _get_api_version(self, optional_params: dict) -> str:
+        """Get API version from optional params or use default."""
+        return optional_params.get("api_version", self.DEFAULT_API_VERSION)
+
+    def get_complete_url(
+        self,
+        api_base: Optional[str],
+        api_key: Optional[str],
+        model: str,
+        optional_params: dict,
+        litellm_params: dict,
+        stream: Optional[bool] = None,
+    ) -> str:
+        """
+        Get the base URL for Azure AI Agent Service.
+        
+        The actual endpoint will vary based on the operation:
+        - /openai/threads for creating threads
+        - /openai/threads/{thread_id}/messages for adding messages
+        - /openai/threads/{thread_id}/runs for creating runs
+        
+        This returns the base URL that will be modified for each operation.
+        """
+        if api_base is None:
+            raise ValueError(
+                "api_base is required for Azure AI Agents. Set it via AZURE_AI_API_BASE env var or api_base parameter."
+            )
+
+        # Remove trailing slash if present
+        api_base = api_base.rstrip("/")
+
+        # Return base URL - actual endpoints will be constructed during request
+        return api_base
+
+    def _get_agent_id(self, model: str, optional_params: dict) -> str:
+        """
+        Get the agent ID from model or optional_params.
+
+        model format: "azure_ai/agents/<agent_id>" or "agents/<agent_id>" or just "<agent_id>"
+        """
+        agent_id = optional_params.get("agent_id") or optional_params.get("assistant_id")
+        if agent_id:
+            return agent_id
+
+        # Extract from model name using the static method
+        return self.get_agent_id_from_model(model)
+
+    def transform_request(
+        self,
+        model: str,
+        messages: List[AllMessageValues],
+        optional_params: dict,
+        litellm_params: dict,
+        headers: dict,
+    ) -> dict:
+        """
+        Transform the request for Azure Agents.
+        
+        This stores the necessary data for the multi-step agent flow.
+        The actual API calls happen in the custom handler.
+        """
+        agent_id = self._get_agent_id(model, optional_params)
+
+        # Convert messages to a format we can use
+        converted_messages = []
+        for msg in messages:
+            role = msg.get("role", "user")
+            content = msg.get("content", "")
+
+            # Handle content that might be a list
+            if isinstance(content, list):
+                content = convert_content_list_to_str(msg)
+
+            # Ensure content is a string
+            if not isinstance(content, str):
+                content = str(content)
+
+            converted_messages.append({"role": role, "content": content})
+
+        payload: Dict[str, Any] = {
+            "agent_id": agent_id,
+            "messages": converted_messages,
+            "api_version": self._get_api_version(optional_params),
+        }
+
+        # Pass through thread_id if provided (for continuing conversations)
+        if "thread_id" in optional_params:
+            payload["thread_id"] = optional_params["thread_id"]
+
+        # Pass through any additional instructions
+        if "instructions" in optional_params:
+            payload["instructions"] = optional_params["instructions"]
+
+        verbose_logger.debug(f"Azure AI Agents request payload: {payload}")
+        return payload
+
+    def validate_environment(
+        self,
+        headers: dict,
+        model: str,
+        messages: List[AllMessageValues],
+        optional_params: dict,
+        litellm_params: dict,
+        api_key: Optional[str] = None,
+        api_base: Optional[str] = None,
+    ) -> dict:
+        """
+        Validate and set up environment for Azure Agents requests.
+        """
+        headers["Content-Type"] = "application/json"
+
+        # Add API key if provided
+        if api_key:
+            headers["api-key"] = api_key
+
+        return headers
+
+    def get_error_class(
+        self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers]
+    ) -> BaseLLMException:
+        return AzureAIAgentsError(status_code=status_code, message=error_message)
+
+    def should_fake_stream(
+        self,
+        model: Optional[str],
+        stream: Optional[bool],
+        custom_llm_provider: Optional[str] = None,
+    ) -> bool:
+        """
+        Azure Agents uses polling, so we fake stream by returning the final response.
+        """
+        return True
+
+    @property
+    def has_custom_stream_wrapper(self) -> bool:
+        """Azure Agents doesn't have native streaming - uses fake stream."""
+        return False
+
+    @property
+    def supports_stream_param_in_request_body(self) -> bool:
+        """
+        Azure Agents does not use a stream param in request body.
+        """
+        return False
+
+    def transform_response(
+        self,
+        model: str,
+        raw_response: httpx.Response,
+        model_response: ModelResponse,
+        logging_obj: LiteLLMLoggingObj,
+        request_data: dict,
+        messages: List[AllMessageValues],
+        optional_params: dict,
+        litellm_params: dict,
+        encoding: Any,
+        api_key: Optional[str] = None,
+        json_mode: Optional[bool] = None,
+    ) -> ModelResponse:
+        """
+        Transform the Azure Agents response to LiteLLM ModelResponse format.
+        """
+        # This is not used since we have a custom handler
+        return model_response
+
+    @staticmethod
+    def completion(
+        model: str,
+        messages: List,
+        api_base: str,
+        api_key: Optional[str],
+        model_response: ModelResponse,
+        logging_obj: LiteLLMLoggingObj,
+        optional_params: dict,
+        litellm_params: dict,
+        timeout: Union[float, int, Any],
+        acompletion: bool,
+        stream: Optional[bool] = False,
+        headers: Optional[dict] = None,
+    ) -> Any:
+        """
+        Dispatch method for Azure AI Agents completion.
+        
+        Routes to sync or async completion based on acompletion flag.
+        Supports native streaming via SSE when stream=True and acompletion=True.
+        """
+        from litellm.llms.azure_ai.agents.handler import azure_ai_agents_handler
+
+        if api_key is None:
+            raise ValueError("api_key is required for Azure AI Agents")
+        if acompletion:
+            if stream:
+                # Native async streaming via SSE - return the async generator directly
+                return azure_ai_agents_handler.acompletion_stream(
+                    model=model,
+                    messages=messages,
+                    api_base=api_base,
+                    api_key=api_key,
+                    logging_obj=logging_obj,
+                    optional_params=optional_params,
+                    litellm_params=litellm_params,
+                    timeout=timeout,
+                    headers=headers,
+                )
+            else:
+                return azure_ai_agents_handler.acompletion(
+                    model=model,
+                    messages=messages,
+                    api_base=api_base,
+                    api_key=api_key,
+                    model_response=model_response,
+                    logging_obj=logging_obj,
+                    optional_params=optional_params,
+                    litellm_params=litellm_params,
+                    timeout=timeout,
+                    headers=headers,
+                )
+        else:
+            # Sync completion - streaming not supported for sync
+            return azure_ai_agents_handler.completion(
+                model=model,
+                messages=messages,
+                api_base=api_base,
+                api_key=api_key,
+                model_response=model_response,
+                logging_obj=logging_obj,
+                optional_params=optional_params,
+                litellm_params=litellm_params,
+                timeout=timeout,
+                headers=headers,
+            )
--- a/litellm/llms/azure_ai/common_utils.py
+++ b/litellm/llms/azure_ai/common_utils.py
@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import List, Literal, Optional

 import litellm
 from litellm.llms.base_llm.base_utils import BaseLLMModelInfo
@ -7,6 +7,17 @@ from litellm.types.llms.openai import AllMessageValues


 class AzureFoundryModelInfo(BaseLLMModelInfo):
+    @staticmethod
+    def get_azure_ai_route(model: str) -> Literal["agents", "default"]:
+        """
+        Get the Azure AI route for the given model.
+        
+        Similar to BedrockModelInfo.get_bedrock_route().
+        """
+        if "agents/" in model:
+            return "agents"
+        return "default"
+
    @staticmethod
    def get_api_base(api_base: Optional[str] = None) -> Optional[str]:
        return (
--- a/litellm/main.py
+++ b/litellm/main.py
@ -1736,9 +1736,37 @@ def completion(  # type: ignore # noqa: PLR0915
        elif custom_llm_provider == "azure_ai":
            from litellm.llms.azure_ai.common_utils import AzureFoundryModelInfo

+            azure_ai_route = AzureFoundryModelInfo.get_azure_ai_route(model)
+
+            # Check if this is an agents route - model format: azure_ai/agents/<agent_id>
+            if azure_ai_route == "agents":
+                from litellm.llms.azure_ai.agents import AzureAIAgentsConfig
+
+                api_base = AzureFoundryModelInfo.get_api_base(api_base)
+                if api_base is None:
+                    raise ValueError(
+                        "Azure AI Agents requests require an api_base. "
+                        "Set `api_base` or the AZURE_AI_API_BASE env var."
+                    )
+                api_key = AzureFoundryModelInfo.get_api_key(api_key)
+
+                response = AzureAIAgentsConfig.completion(
+                    model=model,
+                    messages=messages,
+                    api_base=api_base,
+                    api_key=api_key,
+                    model_response=model_response,
+                    logging_obj=logging,
+                    optional_params=optional_params,
+                    litellm_params=litellm_params,
+                    timeout=timeout,
+                    acompletion=acompletion,
+                    stream=stream,
+                    headers=headers or litellm.headers,
+                )
+
            # Check if this is a Claude model - route to Azure Anthropic handler
-            model_lower = model.lower()
-            if "claude" in model_lower:
+            elif "claude" in model.lower():
                # Use Azure Anthropic handler for Claude models
                api_base = AzureFoundryModelInfo.get_api_base(api_base)
                if api_base is None:
--- a/provider_endpoints_support.json
+++ b/provider_endpoints_support.json
@ -239,6 +239,23 @@
        "ocr": true
      }
    },
+    "azure_ai/agents": {
+      "display_name": "Azure AI Foundry Agents (`azure_ai/agents`)",
+      "url": "https://docs.litellm.ai/docs/providers/azure_ai_agents",
+      "endpoints": {
+        "chat_completions": true,
+        "messages": true,
+        "responses": true,
+        "embeddings": false,
+        "image_generations": false,
+        "audio_transcriptions": false,
+        "audio_speech": false,
+        "moderations": false,
+        "batches": false,
+        "rerank": false,
+        "a2a": true
+      }
+    },
    "azure_text": {
      "display_name": "Azure Text (`azure_text`)",
      "url": "https://docs.litellm.ai/docs/providers/azure",
--- a/tests/llm_translation/test_azure_agents.py
+++ b/tests/llm_translation/test_azure_agents.py
@ -0,0 +1,383 @@
+"""
+Tests for Azure AI Agent Service integration.
+
+These tests require an Azure AI Agent Service endpoint and a pre-configured agent.
+
+The Azure AI Agent Service uses the Assistants API pattern:
+1. Create a thread
+2. Add messages to the thread
+3. Create and poll a run
+4. Get the agent's response messages
+
+Model format: azure_ai/agents/<agent_id>
+
+Example environment variables:
+  AZURE_AI_API_BASE=https://your-project.services.ai.azure.com
+  AZURE_AI_API_KEY=your-api-key
+"""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.abspath("../.."))
+
+import pytest
+
+import litellm
+
+
+@pytest.mark.asyncio
+async def test_azure_ai_agents_acompletion_non_streaming():
+    """
+    Test non-streaming acompletion call to Azure AI Agent Service.
+    Uses the multi-step flow: create thread -> add messages -> create/poll run -> get messages
+    """
+    api_base = os.environ.get("AZURE_API_BASE")
+    api_key = os.environ.get("AZURE_API_KEY")
+    agent_id = "asst_shNRIVxMPuvSRVWP5WvVe4jE"
+
+
+    response = await litellm.acompletion(
+        model=f"azure_ai/agents/{agent_id}",
+        messages=[{"role": "user", "content": "Hi Agent, what is 25 * 4?"}],
+        api_base=api_base,
+        api_key=api_key,
+        stream=False,
+    )
+
+    assert response is not None
+    assert response.choices is not None
+    assert len(response.choices) > 0
+    assert response.choices[0].message is not None
+    assert response.choices[0].message.content is not None
+    assert len(response.choices[0].message.content) > 0
+
+    # Verify thread_id is returned for conversation continuity
+    if hasattr(response, "_hidden_params") and response._hidden_params:
+        assert "thread_id" in response._hidden_params
+
+    print(f"Response: {response.choices[0].message.content}")
+
+
+@pytest.mark.asyncio
+async def test_azure_ai_agents_acompletion_streaming():
+    """
+    Test native streaming acompletion call to Azure AI Agent Service.
+    Uses the create-thread-and-run endpoint with stream=True for SSE streaming.
+    """
+    api_base = os.environ.get("AZURE_API_BASE")
+    api_key = os.environ.get("AZURE_API_KEY")
+    agent_id = os.environ.get("AZURE_AGENTS_AGENT_ID", "asst_shNRIVxMPuvSRVWP5WvVe4jE")
+
+    response = await litellm.acompletion(
+        model=f"azure_ai/agents/{agent_id}",
+        messages=[{"role": "user", "content": "Hi Agent, what is 10 + 5?"}],
+        api_base=api_base,
+        api_key=api_key,
+        stream=True,
+    )
+
+    # Native streaming - collect chunks from the async iterator
+    chunks = []
+    full_content = ""
+    async for chunk in response:
+        print("Streaming chunk: ", chunk)
+        chunks.append(chunk)
+        if hasattr(chunk, "choices") and chunk.choices:
+            delta = chunk.choices[0].delta
+            if hasattr(delta, "content") and delta.content:
+                full_content += delta.content
+
+    assert len(chunks) > 0, "Expected at least one streaming chunk"
+    assert len(full_content) > 0, "Expected content from streaming response"
+    print(f"Streamed response ({len(chunks)} chunks): {full_content}")
+
+
+
+def test_azure_ai_agents_is_agents_route():
+    """
+    Test the is_azure_ai_agents_route detection method.
+    """
+    from litellm.llms.azure_ai.agents.transformation import AzureAIAgentsConfig
+
+    # Should be recognized as agents route
+    assert AzureAIAgentsConfig.is_azure_ai_agents_route("azure_ai/agents/asst_123") is True
+    assert AzureAIAgentsConfig.is_azure_ai_agents_route("agents/asst_123") is True
+    
+    # Should NOT be recognized as agents route
+    assert AzureAIAgentsConfig.is_azure_ai_agents_route("azure_ai/gpt-4") is False
+    assert AzureAIAgentsConfig.is_azure_ai_agents_route("gpt-4") is False
+
+
+def test_azure_ai_get_azure_ai_route():
+    """
+    Test the get_azure_ai_route dispatch method.
+    """
+    from litellm.llms.azure_ai.common_utils import AzureFoundryModelInfo
+
+    # Should return "agents" for agents routes
+    assert AzureFoundryModelInfo.get_azure_ai_route("agents/asst_123") == "agents"
+    assert AzureFoundryModelInfo.get_azure_ai_route("azure_ai/agents/asst_abc") == "agents"
+    
+    # Should return "default" for non-agents routes
+    assert AzureFoundryModelInfo.get_azure_ai_route("gpt-4") == "default"
+    assert AzureFoundryModelInfo.get_azure_ai_route("claude-3-sonnet") == "default"
+    assert AzureFoundryModelInfo.get_azure_ai_route("azure_ai/gpt-4o") == "default"
+
+
+def test_azure_ai_agents_get_agent_id_from_model():
+    """
+    Test agent ID extraction from model name.
+    """
+    from litellm.llms.azure_ai.agents.transformation import AzureAIAgentsConfig
+
+    # Test with full model name
+    agent_id = AzureAIAgentsConfig.get_agent_id_from_model("azure_ai/agents/asst_abc123")
+    assert agent_id == "asst_abc123"
+
+    # Test with just agents/id
+    agent_id = AzureAIAgentsConfig.get_agent_id_from_model("agents/asst_xyz789")
+    assert agent_id == "asst_xyz789"
+
+    # Test with just agent ID (fallback)
+    agent_id = AzureAIAgentsConfig.get_agent_id_from_model("asst_plain")
+    assert agent_id == "asst_plain"
+
+
+def test_azure_ai_agents_config_get_agent_id():
+    """
+    Test agent ID extraction via config method.
+    """
+    from litellm.llms.azure_ai.agents.transformation import AzureAIAgentsConfig
+
+    config = AzureAIAgentsConfig()
+
+    # Test with full model name
+    agent_id = config._get_agent_id("azure_ai/agents/asst_abc123", {})
+    assert agent_id == "asst_abc123"
+
+    # Test with optional_params override
+    agent_id = config._get_agent_id("azure_ai/agents/asst_abc123", {"agent_id": "asst_override"})
+    assert agent_id == "asst_override"
+
+    # Test with assistant_id in optional_params
+    agent_id = config._get_agent_id("azure_ai/agents/asst_abc123", {"assistant_id": "asst_assistant"})
+    assert agent_id == "asst_assistant"
+
+
+def test_azure_ai_agents_config_get_complete_url():
+    """
+    Test that AzureAIAgentsConfig correctly generates base URLs.
+    """
+    from litellm.llms.azure_ai.agents.transformation import AzureAIAgentsConfig
+
+    config = AzureAIAgentsConfig()
+
+    # Test URL generation
+    url = config.get_complete_url(
+        api_base="https://test-project.services.ai.azure.com",
+        api_key=None,
+        model="agents/asst_123",
+        optional_params={},
+        litellm_params={},
+        stream=False,
+    )
+    assert url == "https://test-project.services.ai.azure.com"
+
+    # Test URL with trailing slash
+    url_with_slash = config.get_complete_url(
+        api_base="https://test-project.services.ai.azure.com/",
+        api_key=None,
+        model="agents/asst_123",
+        optional_params={},
+        litellm_params={},
+        stream=False,
+    )
+    assert url_with_slash == "https://test-project.services.ai.azure.com"
+
+
+def test_azure_ai_agents_config_transform_request():
+    """
+    Test that AzureAIAgentsConfig correctly transforms requests.
+    """
+    from litellm.llms.azure_ai.agents.transformation import AzureAIAgentsConfig
+
+    config = AzureAIAgentsConfig()
+
+    messages = [
+        {"role": "system", "content": "You are a helpful assistant."},
+        {"role": "user", "content": "What is 2 + 2?"},
+    ]
+
+    request = config.transform_request(
+        model="azure_ai/agents/asst_123",
+        messages=messages,
+        optional_params={},
+        litellm_params={"stream": False},
+        headers={},
+    )
+
+    assert request["agent_id"] == "asst_123"
+    assert "messages" in request
+    assert len(request["messages"]) == 2
+    assert request["messages"][0]["role"] == "system"
+    assert request["messages"][1]["role"] == "user"
+    assert "api_version" in request
+    assert request["api_version"] == "2024-07-01-preview"
+
+
+def test_azure_ai_agents_provider_detection():
+    """
+    Test that the azure_ai provider is correctly detected from model name.
+    """
+    from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
+
+    model, provider, api_key, api_base = get_llm_provider(
+        model="azure_ai/agents/asst_abc123",
+        api_base="https://test.services.ai.azure.com",
+    )
+
+    assert provider == "azure_ai"
+    assert model == "agents/asst_abc123"
+
+
+def test_azure_ai_agents_validate_environment():
+    """
+    Test that headers are correctly set up.
+    """
+    from litellm.llms.azure_ai.agents.transformation import AzureAIAgentsConfig
+
+    config = AzureAIAgentsConfig()
+
+    headers = config.validate_environment(
+        headers={},
+        model="agents/asst_123",
+        messages=[],
+        optional_params={},
+        litellm_params={},
+        api_key="test-api-key",
+        api_base="https://test.services.ai.azure.com",
+    )
+
+    assert headers["Content-Type"] == "application/json"
+    assert headers["api-key"] == "test-api-key"
+
+
+def test_azure_ai_agents_handler_url_builders():
+    """
+    Test the URL building methods in the handler.
+    """
+    from litellm.llms.azure_ai.agents.handler import AzureAIAgentsHandler
+
+    handler = AzureAIAgentsHandler()
+    api_base = "https://test.services.ai.azure.com"
+    api_version = "2024-07-01-preview"
+    thread_id = "thread_abc123"
+    run_id = "run_xyz789"
+
+    # Test thread URL - uses /openai/ prefix
+    thread_url = handler._build_thread_url(api_base, api_version)
+    assert thread_url == f"{api_base}/openai/threads?api-version={api_version}"
+
+    # Test messages URL
+    messages_url = handler._build_messages_url(api_base, thread_id, api_version)
+    assert messages_url == f"{api_base}/openai/threads/{thread_id}/messages?api-version={api_version}"
+
+    # Test runs URL
+    runs_url = handler._build_runs_url(api_base, thread_id, api_version)
+    assert runs_url == f"{api_base}/openai/threads/{thread_id}/runs?api-version={api_version}"
+
+    # Test run status URL
+    status_url = handler._build_run_status_url(api_base, thread_id, run_id, api_version)
+    assert status_url == f"{api_base}/openai/threads/{thread_id}/runs/{run_id}?api-version={api_version}"
+
+
+def test_azure_ai_agents_extract_content_from_messages():
+    """
+    Test content extraction from Azure Agents message response.
+    """
+    from litellm.llms.azure_ai.agents.handler import AzureAIAgentsHandler
+
+    handler = AzureAIAgentsHandler()
+
+    # Test typical message response
+    messages_data = {
+        "data": [
+            {
+                "id": "msg_123",
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "text",
+                        "text": {"value": "The answer is 100."}
+                    }
+                ]
+            },
+            {
+                "id": "msg_122",
+                "role": "user",
+                "content": [
+                    {
+                        "type": "text",
+                        "text": {"value": "What is 25 * 4?"}
+                    }
+                ]
+            }
+        ]
+    }
+
+    content = handler._extract_content_from_messages(messages_data)
+    assert content == "The answer is 100."
+
+    # Test empty response
+    empty_data = {"data": []}
+    content = handler._extract_content_from_messages(empty_data)
+    assert content == ""
+
+
+@pytest.mark.asyncio
+async def test_azure_ai_agents_conversation_continuity():
+    """
+    Test that thread_id can be used for conversation continuity.
+    """
+    api_base = os.environ.get("AZURE_AI_API_BASE")
+    api_key = os.environ.get("AZURE_AI_API_KEY")
+    agent_id = os.environ.get("AZURE_AI_AGENTS_AGENT_ID", "asst_shNRIVxMPuvSRVWP5WvVe4jE")
+
+    if not api_base or not api_key:
+        pytest.skip("AZURE_AI_API_BASE and AZURE_AI_API_KEY environment variables required")
+
+    try:
+        # First message
+        response1 = await litellm.acompletion(
+            model=f"azure_ai/agents/{agent_id}",
+            messages=[{"role": "user", "content": "My name is Alice. Remember this."}],
+            api_base=api_base,
+            api_key=api_key,
+            stream=False,
+        )
+
+        assert response1 is not None
+        
+        # Get thread_id for continuity
+        thread_id = None
+        if hasattr(response1, "_hidden_params") and response1._hidden_params:
+            thread_id = response1._hidden_params.get("thread_id")
+        
+        if thread_id:
+            # Second message using the same thread
+            response2 = await litellm.acompletion(
+                model=f"azure_ai/agents/{agent_id}",
+                messages=[{"role": "user", "content": "What is my name?"}],
+                api_base=api_base,
+                api_key=api_key,
+                thread_id=thread_id,  # Continue the conversation
+                stream=False,
+            )
+
+            assert response2 is not None
+            # The agent should remember the name from the previous message
+            print(f"Response to name question: {response2.choices[0].message.content}")
+
+    except Exception as e:
+        pytest.skip(f"Azure Agent Service not available: {e}")