Add watson agent

examples/watsonx_agent_a2a_config.yaml

@@ -0,0 +1,82 @@
+# Example configuration for using WatsonX Agents with A2A Protocol
+#
+# This configuration demonstrates how to expose WatsonX agents through
+# LiteLLM's A2A Agent Gateway using the completion bridge adapter.
+#
+# Usage:
+#   1. Set environment variables:
+#      export WATSONX_API_KEY="your-api-key"
+#      export WATSONX_API_BASE="https://your-watsonx-endpoint.com"
+#   
+#   2. Start the proxy:
+#      litellm --config watsonx_agent_a2a_config.yaml
+#   
+#   3. Invoke via A2A protocol:
+#      curl -X POST http://localhost:4000/a2a/watsonx-assistant/message/send \
+#        -H "Authorization: Bearer sk-1234" \
+#        -H "Content-Type: application/json" \
+#        -d '{"jsonrpc":"2.0","id":"req-1","method":"message/send","params":{"message":{"role":"user","parts":[{"kind":"text","text":"Hello!"}],"messageId":"msg-1"}}}'
+
+model_list:
+  # Standard completion endpoint (OpenAI-compatible)
+  - model_name: watsonx-assistant-completion
+    litellm_params:
+      model: watsonx_agent/your-agent-id-here
+      api_base: os.environ/WATSONX_API_BASE
+      api_key: os.environ/WATSONX_API_KEY
+
+# Agents configuration for A2A Protocol
+agents:
+  # Customer support agent
+  - agent_id: watsonx-assistant
+    agent_name: WatsonX Assistant
+    litellm_params:
+      custom_llm_provider: watsonx           # Required: tells LiteLLM to use watsonx provider
+      model: watsonx_agent/your-agent-id-here  # WatsonX agent ID
+      api_key: os.environ/WATSONX_API_KEY    # WatsonX API key
+    agent_card_params:
+      name: WatsonX Assistant
+      description: AI assistant powered by IBM WatsonX Orchestrate
+      url: os.environ/WATSONX_API_BASE       # WatsonX API endpoint
+      capabilities:
+        - text_generation
+        - conversation
+        - multi_turn_dialogue
+      version: "1.0"
+
+  # Technical support agent (example with multiple agents)
+  - agent_id: watsonx-tech-support
+    agent_name: Technical Support Agent
+    litellm_params:
+      custom_llm_provider: watsonx
+      model: watsonx_agent/tech-support-agent-id
+      api_key: os.environ/WATSONX_API_KEY
+    agent_card_params:
+      name: Technical Support Agent
+      description: Specialized agent for technical troubleshooting
+      url: os.environ/WATSONX_API_BASE
+      capabilities:
+        - text_generation
+        - conversation
+        - technical_support
+
+# Optional: General settings
+general_settings:
+  master_key: sk-1234  # Change this! Used to authenticate requests
+  
+  # Enable detailed logging
+  set_verbose: true
+  
+  # Cost tracking
+  track_cost_per_deployment: true
+
+# Optional: Team-based access control
+# Only users in specified teams can access certain agents
+router_settings:
+  enable_pre_call_checks: true
+
+# Optional: Rate limiting
+# litellm_settings:
+#   num_retries: 3
+#   request_timeout: 600
+#   max_budget: 100  # USD per month

litellm/llms/watsonx/agents/A2A_INTEGRATION.md

@@ -0,0 +1,356 @@
+# WatsonX Agents with A2A Protocol (Agent Gateway)
+
+WatsonX agents can be exposed through LiteLLM's A2A Agent Gateway using the **completion bridge** adapter. This allows you to invoke WatsonX agents using the A2A JSON-RPC protocol while maintaining all LiteLLM features like logging, cost tracking, and access control.
+
+## How It Works
+
+The completion bridge acts as an adapter that:
+
+1. **Receives A2A JSON-RPC requests** (with messages in A2A format)
+2. **Transforms to OpenAI format** using transformation utilities
+3. **Routes through `litellm.completion()`** with `watsonx_agent` provider
+4. **Converts responses back to A2A JSON-RPC format**
+
+This means WatsonX agents work with the A2A protocol **without requiring native A2A support** in the WatsonX API itself.
+
+## Architecture
+
+```
+A2A Client → LiteLLM A2A Endpoint → Completion Bridge → WatsonX Agent Handler → WatsonX API
+   (A2A)         (A2A ↔ OpenAI)        (OpenAI ↔ WatsonX)
+```
+
+## Configuration
+
+### Option 1: LiteLLM Proxy Configuration
+
+Add your WatsonX agent to `proxy_server_config.yaml`:
+
+```yaml
+model_list:
+  - model_name: my-watsonx-agent
+    litellm_params:
+      model: watsonx_agent/your-agent-id
+      api_base: https://your-watsonx-endpoint.com
+      api_key: os.environ/WATSONX_API_KEY
+
+# Register agent for A2A access
+agents:
+  - agent_id: watsonx-assistant
+    agent_name: WatsonX Assistant
+    litellm_params:
+      custom_llm_provider: watsonx
+      model: watsonx_agent/your-agent-id
+      api_key: os.environ/WATSONX_API_KEY
+    agent_card_params:
+      name: WatsonX Assistant
+      description: AI assistant powered by IBM WatsonX
+      url: https://your-watsonx-endpoint.com
+```
+
+Start the proxy:
+
+```bash
+litellm --config proxy_server_config.yaml
+```
+
+### Option 2: Direct Python SDK Usage
+
+```python
+from litellm.a2a_protocol import asend_message
+from a2a.types import SendMessageRequest, MessageSendParams
+from uuid import uuid4
+
+# Create A2A request
+request = SendMessageRequest(
+    id=str(uuid4()),
+    params=MessageSendParams(
+        message={
+            "role": "user",
+            "parts": [{"kind": "text", "text": "Hello, how can you help me?"}],
+            "messageId": uuid4().hex,
+        }
+    ),
+)
+
+# Send via completion bridge
+response = await asend_message(
+    request=request,
+    api_base="https://your-watsonx-endpoint.com",
+    litellm_params={
+        "custom_llm_provider": "watsonx",
+        "model": "watsonx_agent/your-agent-id",
+        "api_key": "your-api-key",
+    },
+)
+
+print(response.result["message"]["parts"][0]["text"])
+```
+
+## Invoking the Agent
+
+### Using A2A SDK
+
+```python
+from a2a.client import A2AClient
+from a2a.types import SendMessageRequest, MessageSendParams
+from uuid import uuid4
+
+# Connect to LiteLLM proxy
+client = await A2AClient.create(base_url="http://localhost:4000/a2a/watsonx-assistant")
+
+# Send message
+request = SendMessageRequest(
+    id=str(uuid4()),
+    params=MessageSendParams(
+        message={
+            "role": "user",
+            "parts": [{"kind": "text", "text": "What can you do?"}],
+            "messageId": uuid4().hex,
+        }
+    ),
+)
+
+response = await client.send_message(request)
+print(response.result.message.parts[0].text)
+```
+
+### Using OpenAI SDK (Alternative)
+
+You can also invoke via the standard OpenAI-compatible `/chat/completions` endpoint:
+
+```python
+import openai
+
+client = openai.OpenAI(
+    base_url="http://localhost:4000",
+    api_key="your-litellm-proxy-key"
+)
+
+response = client.chat.completions.create(
+    model="a2a/watsonx-assistant",  # Note: a2a/ prefix
+    messages=[{"role": "user", "content": "Hello!"}]
+)
+
+print(response.choices[0].message.content)
+```
+
+### Using HTTP/cURL
+
+```bash
+# A2A Protocol endpoint
+curl -X POST http://localhost:4000/a2a/watsonx-assistant/message/send \
+  -H "Authorization: Bearer sk-your-litellm-key" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "jsonrpc": "2.0",
+    "id": "req-123",
+    "method": "message/send",
+    "params": {
+      "message": {
+        "role": "user",
+        "parts": [{"kind": "text", "text": "Hello!"}],
+        "messageId": "msg-123"
+      }
+    }
+  }'
+```
+
+## Streaming Support
+
+WatsonX agents support streaming through the A2A protocol:
+
+```python
+from litellm.a2a_protocol import asend_message_streaming
+from a2a.types import SendStreamingMessageRequest, MessageSendParams
+from uuid import uuid4
+
+request = SendStreamingMessageRequest(
+    id=str(uuid4()),
+    params=MessageSendParams(
+        message={
+            "role": "user",
+            "parts": [{"kind": "text", "text": "Tell me a story"}],
+            "messageId": uuid4().hex,
+        }
+    ),
+)
+
+async for chunk in asend_message_streaming(
+    request=request,
+    api_base="https://your-watsonx-endpoint.com",
+    litellm_params={
+        "custom_llm_provider": "watsonx",
+        "model": "watsonx_agent/your-agent-id",
+        "api_key": "your-api-key",
+    },
+):
+    print(chunk)
+```
+
+The streaming response follows the A2A protocol with these events:
+1. **Task event** (`kind: "task"`) - Initial task creation
+2. **Status update** (`kind: "status-update"`) - Status change to "working"
+3. **Artifact update** (`kind: "artifact-update"`) - Content delivery
+4. **Status update** (`kind: "status-update"`) - Final "completed" status
+
+## Thread Continuity
+
+WatsonX agents support conversation continuity through thread IDs:
+
+```python
+# First message - creates a new thread
+response1 = await asend_message(
+    request=request1,
+    api_base=api_base,
+    litellm_params={
+        "custom_llm_provider": "watsonx",
+        "model": "watsonx_agent/your-agent-id",
+        "api_key": api_key,
+    },
+)
+
+# Get thread_id from response
+thread_id = response1._hidden_params.get("thread_id")
+
+# Continue conversation with same thread
+response2 = await asend_message(
+    request=request2,
+    api_base=api_base,
+    litellm_params={
+        "custom_llm_provider": "watsonx",
+        "model": "watsonx_agent/your-agent-id",
+        "api_key": api_key,
+        "thread_id": thread_id,  # Continue same conversation
+    },
+)
+```
+
+## Features
+
+| Feature | Supported |
+|---------|-----------|
+| **A2A Protocol** | ✅ (via completion bridge) |
+| **Streaming** | ✅ |
+| **Thread Continuity** | ✅ |
+| **Cost Tracking** | ✅ |
+| **Logging** | ✅ |
+| **Access Control** | ✅ |
+| **Load Balancing** | ✅ |
+
+## Benefits of A2A Integration
+
+1. **Standardized Protocol**: Use the same A2A protocol across different agent providers (WatsonX, Vertex AI, Azure AI, LangGraph, etc.)
+2. **Cost Tracking**: Automatic cost calculation and logging for WatsonX agent usage
+3. **Access Control**: Team-based and key-based access control to agents
+4. **Observability**: Unified logging across all agent calls through LiteLLM
+5. **Discovery**: Agents are discoverable through the AI Hub registry
+
+## Comparison with Other Providers
+
+| Provider | A2A Support | Implementation |
+|----------|-------------|----------------|
+| Native A2A Agents | ✅ | Direct A2A protocol |
+| Vertex AI Agent Engine | ✅ | Completion bridge |
+| Azure AI Foundry | ✅ | Completion bridge |
+| LangGraph | ✅ | Completion bridge |
+| Bedrock AgentCore | ✅ | Completion bridge |
+| **WatsonX Agents** | ✅ | **Completion bridge** |
+
+## Authentication
+
+WatsonX agents support multiple authentication methods:
+
+1. **Bearer Token**:
+   ```python
+   litellm_params={"api_key": "Bearer your-token"}
+   ```
+
+2. **IAM API Key** (automatically exchanges for token):
+   ```python
+   litellm_params={"api_key": "your-iam-api-key"}
+   ```
+
+3. **ZenApiKey** (for IBM Cloud Pak for Data):
+   ```python
+   litellm_params={"zen_api_key": "your-zen-api-key"}
+   ```
+
+## Example: Full Integration
+
+Here's a complete example using WatsonX agents with the LiteLLM proxy:
+
+```yaml
+# proxy_server_config.yaml
+model_list:
+  - model_name: customer-support-agent
+    litellm_params:
+      model: watsonx_agent/agent-123
+      api_base: https://watsonx.example.com
+      api_key: os.environ/WATSONX_API_KEY
+
+agents:
+  - agent_id: customer-support
+    agent_name: Customer Support Agent
+    litellm_params:
+      custom_llm_provider: watsonx
+      model: watsonx_agent/agent-123
+      api_key: os.environ/WATSONX_API_KEY
+    agent_card_params:
+      name: Customer Support Agent
+      description: Handles customer inquiries and support tickets
+      url: https://watsonx.example.com
+      capabilities:
+        - text_generation
+        - conversation
+```
+
+Client code:
+
+```python
+from a2a.client import A2AClient
+
+# Connect to agent via LiteLLM proxy
+client = await A2AClient.create(
+    base_url="http://localhost:4000/a2a/customer-support",
+    headers={"Authorization": "Bearer sk-your-litellm-key"}
+)
+
+# Use the agent
+response = await client.send_message(request)
+```
+
+## Troubleshooting
+
+### Agent not found
+- Ensure the agent is registered in `agents:` section of config
+- Check that `agent_id` matches the URL path: `/a2a/{agent_id}`
+
+### Authentication errors
+- Verify `WATSONX_API_KEY` is set correctly
+- Check that the API key has access to the specified agent
+- For IAM keys, ensure they're not expired
+
+### Response format issues
+- The completion bridge automatically handles format conversion
+- Check LiteLLM logs for transformation errors: `litellm --debug`
+
+## Testing
+
+Run the test suite:
+
+```bash
+# Set environment variables
+export WATSONX_API_BASE="https://your-endpoint.com"
+export WATSONX_API_KEY="your-api-key"
+export WATSONX_AGENT_ID="your-agent-id"
+
+# Run tests
+pytest tests/agent_tests/local_only_agent_tests/test_a2a_watsonx_agent.py -v -s
+```
+
+## References
+
+- [LiteLLM Agent Gateway Documentation](https://docs.litellm.ai/docs/a2a)
+- [WatsonX Agents Documentation](./README.md)
+- [A2A Protocol Specification](https://github.com/google/a2a)

litellm/llms/watsonx/agents/README.md

@@ -0,0 +1,343 @@
+# IBM watsonx.ai Orchestrate Agent Support
+
+This module provides support for IBM watsonx.ai Orchestrate Agents in LiteLLM, following the same pattern as Azure AI agents and other agent implementations.
+
+## Model Format
+
+```
+watsonx_agent/<agent_id>
+```
+
+## API Documentation
+
+- [Chat With Agents API](https://developer.watson-orchestrate.ibm.com/apis/orchestrate-agent/chat-with-agents)
+
+## Usage
+
+### Basic Example
+
+```python
+import litellm
+
+response = litellm.completion(
+    model="watsonx_agent/your-agent-id",
+    messages=[
+        {"role": "user", "content": "Hello, how can you help me?"}
+    ],
+    api_base="https://your-watsonx-api-endpoint.com",
+    api_key="your-api-key"
+)
+
+print(response.choices[0].message.content)
+```
+
+### With Thread ID for Conversation Continuity
+
+```python
+import litellm
+
+# First message - creates a new thread
+response = litellm.completion(
+    model="watsonx_agent/your-agent-id",
+    messages=[
+        {"role": "user", "content": "What's the weather like today?"}
+    ],
+    api_base="https://your-watsonx-api-endpoint.com",
+    api_key="your-api-key"
+)
+
+# Get thread_id from response
+thread_id = response._hidden_params.get("thread_id")
+print(f"Thread ID: {thread_id}")
+
+# Continue conversation with the same thread
+response2 = litellm.completion(
+    model="watsonx_agent/your-agent-id",
+    messages=[
+        {"role": "user", "content": "What about tomorrow?"}
+    ],
+    api_base="https://your-watsonx-api-endpoint.com",
+    api_key="your-api-key",
+    thread_id=thread_id  # Continue same conversation
+)
+```
+
+### With Additional Parameters and Context
+
+```python
+import litellm
+
+response = litellm.completion(
+    model="watsonx_agent/your-agent-id",
+    messages=[
+        {"role": "user", "content": "Help me with my task"}
+    ],
+    api_base="https://your-watsonx-api-endpoint.com",
+    api_key="your-api-key",
+    additional_parameters={
+        "custom_param": "value"
+    },
+    context={
+        "user_id": "123",
+        "session_info": "relevant context"
+    }
+)
+```
+
+### Async Example
+
+```python
+import asyncio
+import litellm
+
+async def main():
+    response = await litellm.acompletion(
+        model="watsonx_agent/your-agent-id",
+        messages=[
+            {"role": "user", "content": "Hello!"}
+        ],
+        api_base="https://your-watsonx-api-endpoint.com",
+        api_key="your-api-key"
+    )
+    print(response.choices[0].message.content)
+
+asyncio.run(main())
+```
+
+## Configuration
+
+### Environment Variables
+
+You can set the following environment variables:
+
+```bash
+export WATSONX_API_BASE="https://your-watsonx-api-endpoint.com"
+export WATSONX_API_KEY="your-api-key"
+```
+
+Then call without explicit parameters:
+
+```python
+import litellm
+import os
+
+# Uses environment variables
+response = litellm.completion(
+    model="watsonx_agent/your-agent-id",
+    messages=[{"role": "user", "content": "Hello!"}]
+)
+```
+
+### Using with LiteLLM Proxy
+
+Add to your `config.yaml`:
+
+```yaml
+model_list:
+  - model_name: my-watsonx-agent
+    litellm_params:
+      model: watsonx_agent/your-agent-id
+      api_base: https://your-watsonx-endpoint.com
+      api_key: os.environ/WATSONX_API_KEY
+```
+
+Then call via the proxy:
+
+```python
+import openai
+
+client = openai.OpenAI(
+    base_url="http://localhost:4000",
+    api_key="your-litellm-proxy-key"
+)
+
+response = client.chat.completions.create(
+    model="my-watsonx-agent",
+    messages=[{"role": "user", "content": "Hello!"}]
+)
+```
+
+## API Parameters
+
+### Required Parameters
+
+- `model`: Model identifier in format `watsonx_agent/{agent_id}`
+- `messages`: List of message dictionaries with `role` and `content`
+- `api_base`: Base URL for the watsonx API endpoint
+- `api_key`: Authentication API key
+
+### Optional Parameters
+
+- `thread_id`: Thread ID to continue a conversation
+- `additional_parameters`: Dictionary of additional parameters
+- `context`: Context dictionary for the agent
+- `stream`: Enable streaming (default: True)
+
+### Response Format
+
+The response follows the standard LiteLLM format:
+
+```python
+{
+    "id": "chatcmpl-123",
+    "object": "chat.completion",
+    "created": 1677652288,
+    "model": "watsonx_agent/abc123",
+    "choices": [
+        {
+            "index": 0,
+            "message": {
+                "role": "assistant",
+                "content": "Response from the agent"
+            },
+            "finish_reason": "stop"
+        }
+    ],
+    "_hidden_params": {
+        "thread_id": "thread-xyz789"  # For conversation continuity
+    }
+}
+```
+
+## Authentication
+
+The watsonx agent API supports multiple authentication methods:
+
+1. **Bearer Token**: Directly provide a token
+   ```python
+   api_key="Bearer your-token"
+   ```
+
+2. **IAM API Key**: Provide an IAM API key (automatically exchanges for a bearer token)
+   ```python
+   api_key="your-iam-api-key"
+   ```
+
+3. **ZenApiKey**: For IBM Cloud Pak for Data environments
+   ```python
+   zen_api_key="your-zen-api-key"
+   ```
+
+## Error Handling
+
+```python
+import litellm
+from litellm.llms.watsonx.common_utils import WatsonXAIError
+
+try:
+    response = litellm.completion(
+        model="watsonx_agent/your-agent-id",
+        messages=[{"role": "user", "content": "Hello"}],
+        api_base="https://your-watsonx-api-endpoint.com",
+        api_key="your-api-key"
+    )
+except WatsonXAIError as e:
+    print(f"WatsonX Error: {e.status_code} - {e.message}")
+except Exception as e:
+    print(f"Error: {str(e)}")
+```
+
+## Architecture
+
+The implementation follows the same pattern as Azure AI agents:
+
+### Files Structure
+
+```
+litellm/llms/watsonx/agents/
+├── __init__.py              # Package initialization
+├── transformation.py         # IBMWatsonXAgentConfig class
+├── handler.py               # WatsonXAgentHandler class
+└── README.md                # This file
+
+litellm/types/llms/
+└── watsonx_agents.py        # Type definitions
+
+tests/test_litellm/llms/watsonx/agents/
+└── test_watsonx_agents_transformation.py  # Tests
+```
+
+### How It Works
+
+1. **Provider Detection**: When you use `model="watsonx_agent/agent-id"`, LiteLLM detects the `watsonx_agent` provider
+2. **Static Dispatch**: Routes to `IBMWatsonXAgentConfig.completion()` static method
+3. **Handler Execution**: Dispatches to `watsonx_agent_handler` singleton for sync or async execution
+4. **Request Transformation**: Converts OpenAI-style messages to watsonx agent format
+5. **API Call**: Makes HTTP request to watsonx Orchestrate Agent API
+6. **Response Transformation**: Converts watsonx response back to OpenAI format
+
+### Comparison with Azure AI Agents
+
+| Aspect | Azure AI Agents | Watsonx Agents |
+|--------|----------------|----------------|
+| **Model Format** | `azure_ai/agents/<agent_id>` | `watsonx_agent/<agent_id>` |
+| **API Flow** | Multi-step (thread → messages → run → poll) | Single API call |
+| **Threading** | Manual thread management | Thread ID in header |
+| **Authentication** | Azure AD Bearer tokens | IAM/Bearer tokens |
+| **Static Dispatch** | ✅ Yes | ✅ Yes |
+| **Async Support** | ✅ Yes | ✅ Yes |
+
+## Features
+
+- ✅ Synchronous and asynchronous completions
+- ✅ Thread-based conversation management
+- ✅ Multiple authentication methods
+- ✅ Custom parameters and context support
+- ✅ Error handling with detailed error messages
+- ✅ Follows LiteLLM agent implementation patterns
+- ✅ Compatible with LiteLLM Proxy
+- ✅ Environment variable configuration
+
+## Testing
+
+Run the tests with:
+
+```bash
+poetry run pytest tests/test_litellm/llms/watsonx/agents/ -v
+```
+
+## Implementation Details
+
+### IBMWatsonXAgentConfig
+
+Configuration class that handles:
+- Parameter mapping
+- URL building
+- Request/response transformation
+- Environment validation
+- Static dispatch to handler
+
+### WatsonXAgentHandler
+
+Handler class that executes:
+- Synchronous completions
+- Asynchronous completions
+- HTTP client management
+- Error handling
+
+## Contributing
+
+When contributing to watsonx agent support:
+
+1. Follow the existing code patterns in the watsonx module
+2. Match the Azure AI agents implementation pattern
+3. Add tests for new features
+4. Update this README with new examples
+5. Ensure all tests pass with `make test-unit`
+
+## Comparison with Other LiteLLM Agent Implementations
+
+### Azure AI Agents
+- Multi-step flow with polling
+- Thread and run management
+- Similar static dispatch pattern
+
+### Langraph Agents
+- Graph-based execution
+- State management
+- Different execution model
+
+### Watsonx Agents (This Implementation)
+- Single API call model
+- Simple thread continuation
+- Follows Azure AI pattern for consistency

litellm/llms/watsonx/agents/__init__.py

@@ -0,0 +1,3 @@
+"""
+IBM watsonx.ai Orchestrate Agent support for LiteLLM.
+"""

litellm/llms/watsonx/agents/handler.py

@@ -0,0 +1,259 @@
+"""
+Handler for IBM watsonx.ai Orchestrate Agent API.
+
+Model format: watsonx_agent/<agent_id>
+
+API Reference: https://developer.watson-orchestrate.ibm.com/apis/orchestrate-agent/chat-with-agents
+"""
+
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+
+import httpx
+
+from litellm._logging import verbose_logger
+from litellm.types.utils import ModelResponse
+
+from .transformation import IBMWatsonXAgentConfig
+
+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 WatsonXAgentHandler:
+    """
+    Handler for watsonx agent completions.
+
+    Executes agent API calls to watsonx Orchestrate.
+    """
+
+    def __init__(self):
+        self.config = IBMWatsonXAgentConfig()
+
+    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,
+        headers: Optional[dict] = None,
+    ) -> ModelResponse:
+        """
+        Execute synchronous completion using watsonx agent API.
+
+        Args:
+            model: Model identifier (format: watsonx_agent/AGENT_ID)
+            messages: List of messages
+            api_base: API base URL
+            api_key: API key
+            model_response: ModelResponse object to populate
+            logging_obj: Logging object
+            optional_params: Optional parameters
+            litellm_params: LiteLLM parameters
+            timeout: Request timeout
+            headers: Request headers
+
+        Returns:
+            ModelResponse object
+        """
+        from litellm.llms.custom_httpx.http_handler import _get_httpx_client
+
+        client = _get_httpx_client(params={"ssl_verify": litellm_params.get("ssl_verify", None)})
+
+        # Validate environment and update headers
+        headers = self.config.validate_environment(
+            headers=headers or {},
+            model=model,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            api_key=api_key,
+            api_base=api_base,
+        )
+
+        # Get complete URL
+        url = self.config.get_complete_url(
+            api_base=api_base,
+            api_key=api_key,
+            model=model,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            stream=optional_params.get("stream", False),
+        )
+
+        # Transform request
+        request_data = self.config.transform_request(
+            model=model,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            headers=headers,
+        )
+
+        # Add thread_id header if provided
+        thread_id = optional_params.get("thread_id")
+        if thread_id:
+            headers["X-IBM-THREAD-ID"] = str(thread_id)
+
+        verbose_logger.debug(
+            f"Watsonx Agent API request - URL: {url}, Headers: {headers}, Data: {request_data}"
+        )
+
+        # Make synchronous request
+        response = client.post(
+            url=url,
+            json=request_data,
+            headers=headers,
+            timeout=timeout,
+        )
+
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as e:
+            error_message = e.response.text
+            status_code = e.response.status_code
+            raise self.config.get_error_class(
+                error_message=error_message,
+                status_code=status_code,
+                headers=e.response.headers,
+            )
+
+        # Transform response
+        return self.config.transform_response(
+            model=model,
+            raw_response=response,
+            model_response=model_response,
+            logging_obj=logging_obj,
+            request_data=request_data,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            encoding=None,
+            api_key=api_key,
+        )
+
+    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,
+        headers: Optional[dict] = None,
+    ) -> ModelResponse:
+        """
+        Execute asynchronous completion using watsonx agent API.
+
+        Args:
+            model: Model identifier (format: watsonx_agent/AGENT_ID)
+            messages: List of messages
+            api_base: API base URL
+            api_key: API key
+            model_response: ModelResponse object to populate
+            logging_obj: Logging object
+            optional_params: Optional parameters
+            litellm_params: LiteLLM parameters
+            timeout: Request timeout
+            headers: Request headers
+
+        Returns:
+            ModelResponse object
+        """
+        import litellm
+        from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
+
+        client = get_async_httpx_client(
+            llm_provider=litellm.LlmProviders.WATSONX,
+            params={"ssl_verify": litellm_params.get("ssl_verify", None)},
+        )
+
+        # Validate environment and update headers
+        headers = self.config.validate_environment(
+            headers=headers or {},
+            model=model,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            api_key=api_key,
+            api_base=api_base,
+        )
+
+        # Get complete URL
+        url = self.config.get_complete_url(
+            api_base=api_base,
+            api_key=api_key,
+            model=model,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            stream=optional_params.get("stream", False),
+        )
+
+        # Transform request
+        request_data = self.config.transform_request(
+            model=model,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            headers=headers,
+        )
+
+        # Add thread_id header if provided
+        thread_id = optional_params.get("thread_id")
+        if thread_id:
+            headers["X-IBM-THREAD-ID"] = str(thread_id)
+
+        verbose_logger.debug(
+            f"Watsonx Agent API request - URL: {url}, Headers: {headers}, Data: {request_data}"
+        )
+
+        # Make asynchronous request
+        response = await client.post(
+            url=url,
+            json=request_data,
+            headers=headers,
+            timeout=timeout,
+        )
+
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as e:
+            error_message = e.response.text
+            status_code = e.response.status_code
+            raise self.config.get_error_class(
+                error_message=error_message,
+                status_code=status_code,
+                headers=e.response.headers,
+            )
+
+        # Transform response
+        return self.config.transform_response(
+            model=model,
+            raw_response=response,
+            model_response=model_response,
+            logging_obj=logging_obj,
+            request_data=request_data,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            encoding=None,
+            api_key=api_key,
+        )
+
+
+# Singleton instance
+watsonx_agent_handler = WatsonXAgentHandler()

litellm/llms/watsonx/agents/transformation.py

@@ -0,0 +1,424 @@
+"""
+Transformation for IBM watsonx.ai Orchestrate Agent API.
+
+Model format: watsonx_agent/<agent_id>
+
+API Reference: https://developer.watson-orchestrate.ibm.com/apis/orchestrate-agent/chat-with-agents
+"""
+
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, 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.llms.watsonx_agents import (
+    WatsonxAgentChoice,
+    WatsonxAgentMessage,
+    WatsonxAgentResponse,
+)
+from litellm.types.utils import Choices, Message, ModelResponse
+
+from ..common_utils import IBMWatsonXMixin, WatsonXAIError
+
+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 IBMWatsonXAgentConfig(IBMWatsonXMixin, BaseConfig):
+    """Configuration for IBM watsonx.ai Orchestrate Agent API."""
+
+    def __init__(self, **kwargs):
+        BaseConfig.__init__(self, **kwargs)
+
+    def get_supported_openai_params(self, model: str) -> List[str]:
+        """
+        Return list of OpenAI parameters supported by watsonx agents.
+
+        Currently, watsonx agents support a limited set of parameters.
+        Most standard OpenAI completion parameters are not directly supported.
+        """
+        return [
+            "stream",  # Streaming support
+            "messages",  # Required messages parameter
+        ]
+
+    def map_openai_params(
+        self,
+        non_default_params: dict,
+        optional_params: dict,
+        model: str,
+        drop_params: bool,
+    ) -> dict:
+        """
+        Map OpenAI parameters to watsonx agent parameters.
+
+        Most OpenAI parameters don't have direct equivalents in watsonx agents,
+        so we'll pass through only what's supported.
+        """
+        # Extract thread_id if provided
+        thread_id = non_default_params.pop("thread_id", None)
+        if thread_id:
+            optional_params["thread_id"] = thread_id
+
+        # Extract additional_parameters if provided
+        additional_parameters = non_default_params.pop("additional_parameters", None)
+        if additional_parameters:
+            optional_params["additional_parameters"] = additional_parameters
+
+        # Extract context if provided
+        context = non_default_params.pop("context", None)
+        if context:
+            optional_params["context"] = context
+
+        return optional_params
+
+    def _get_agent_id(self, model: str) -> str:
+        """
+        Extract agent_id from model string.
+
+        Expected format: "watsonx_agent/{agent_id}"
+        Example: "watsonx_agent/abc123"
+        """
+        if "/" in model:
+            parts = model.split("/")
+            if len(parts) >= 2:
+                return parts[1]
+        raise ValueError(
+            f"Invalid model format for watsonx agent: {model}. "
+            "Expected format: 'watsonx_agent/AGENT_ID'"
+        )
+
+    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:
+        """
+        Build the complete URL for the watsonx agent API.
+
+        URL format: https://{api_endpoint}/api/v1/orchestrate/{agent_id}/chat/completions
+        """
+        base_url = self._get_base_url(api_base=api_base)
+        agent_id = self._get_agent_id(model)
+
+        url = f"{base_url.rstrip('/')}/api/v1/orchestrate/{agent_id}/chat/completions"
+
+        verbose_logger.debug(f"Watsonx Agent API URL: {url}")
+        return url
+
+    def _transform_messages(
+        self, messages: List[AllMessageValues]
+    ) -> List[WatsonxAgentMessage]:
+        """
+        Transform OpenAI-style messages to watsonx agent format.
+
+        Args:
+            messages: List of OpenAI-style message dictionaries
+
+        Returns:
+            List of watsonx agent message dictionaries
+        """
+        transformed_messages: List[WatsonxAgentMessage] = []
+
+        for msg in messages:
+            role = msg.get("role", "user")
+            content = msg.get("content", "")
+
+            # Convert content to string if it's a list
+            if isinstance(content, list):
+                content = convert_content_list_to_str(msg)
+
+            transformed_msg: WatsonxAgentMessage = {
+                "role": role,
+                "content": content,  # type: ignore
+            }
+            transformed_messages.append(transformed_msg)
+
+        return transformed_messages
+
+    def transform_request(
+        self,
+        model: str,
+        messages: List[AllMessageValues],
+        optional_params: dict,
+        litellm_params: dict,
+        headers: dict,
+    ) -> dict:
+        """
+        Transform LiteLLM request to watsonx agent API format.
+
+        Args:
+            model: Model identifier
+            messages: List of messages
+            optional_params: Optional parameters
+            litellm_params: LiteLLM parameters
+            headers: Request headers
+
+        Returns:
+            Request body for watsonx agent API
+        """
+        # Transform messages to agent format
+        agent_messages = self._transform_messages(messages)
+
+        # Build request body
+        request_body: dict = {
+            "messages": agent_messages,
+            "additional_parameters": optional_params.get("additional_parameters", {}),
+            "context": optional_params.get("context", {}),
+            "stream": optional_params.get("stream", True),
+        }
+
+        verbose_logger.debug(
+            f"Watsonx Agent request body: {request_body}"
+        )
+        return request_body
+
+    def _transform_agent_choice_to_litellm(
+        self, agent_choice: WatsonxAgentChoice, index: int
+    ) -> Choices:
+        """
+        Transform a watsonx agent choice to LiteLLM format.
+
+        Args:
+            agent_choice: Agent response choice
+            index: Choice index
+
+        Returns:
+            LiteLLM Choices object
+        """
+        agent_message = agent_choice.get("message") or {}
+        content = agent_message.get("content", "")
+
+        # Handle content that might be a list or dict
+        if isinstance(content, (list, dict)):
+            import json
+
+            content = json.dumps(content)
+
+        message = Message(
+            content=str(content),
+            role=agent_message.get("role", "assistant"),
+        )
+
+        finish_reason = agent_choice.get("finish_reason") or "stop"
+
+        return Choices(
+            finish_reason=finish_reason,
+            index=index,
+            message=message,
+        )
+
+    def transform_response(
+        self,
+        model: str,
+        raw_response: httpx.Response,
+        model_response: ModelResponse,
+        logging_obj: Any,
+        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 watsonx agent API response to LiteLLM format.
+
+        Args:
+            model: Model identifier
+            raw_response: Raw HTTP response
+            model_response: LiteLLM ModelResponse object to populate
+            logging_obj: Logging object
+            request_data: Original request data
+            messages: Original messages
+            optional_params: Optional parameters
+            litellm_params: LiteLLM parameters
+            encoding: Encoding
+            api_key: API key
+            json_mode: JSON mode flag
+
+        Returns:
+            Populated ModelResponse object
+        """
+        try:
+            response_json: WatsonxAgentResponse = raw_response.json()
+            verbose_logger.debug(f"Watsonx Agent response: {response_json}")
+
+            # Extract response fields
+            response_id = response_json.get("id", "")
+            created = response_json.get("created", 0)
+            response_model = response_json.get("model", model)
+            thread_id = response_json.get("thread_id", "")
+
+            # Transform choices
+            agent_choices = response_json.get("choices", [])
+            litellm_choices = [
+                self._transform_agent_choice_to_litellm(choice, idx)
+                for idx, choice in enumerate(agent_choices)
+            ]
+
+            # Update model response
+            model_response.id = response_id
+            model_response.created = created
+            model_response.model = response_model
+            model_response.choices = litellm_choices
+
+            # Add thread_id as metadata
+            if thread_id:
+                model_response._hidden_params = model_response._hidden_params or {}
+                model_response._hidden_params["thread_id"] = thread_id
+
+            return model_response
+
+        except Exception as e:
+            verbose_logger.error(
+                f"Error transforming watsonx agent response: {str(e)}"
+            )
+            raise WatsonXAIError(
+                status_code=raw_response.status_code,
+                message=f"Error transforming response: {str(e)}",
+            )
+
+    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 environment and set up authentication headers.
+
+        Args:
+            headers: Request headers
+            model: Model identifier
+            messages: Messages
+            optional_params: Optional parameters
+            litellm_params: LiteLLM parameters
+            api_key: API key
+            api_base: API base URL
+
+        Returns:
+            Updated headers with authentication
+        """
+        # Use the parent class method to validate and set up auth
+        return super().validate_environment(
+            headers=headers,
+            model=model,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            api_key=api_key,
+            api_base=api_base,
+        )
+
+    def get_error_class(
+        self,
+        error_message: str,
+        status_code: int,
+        headers: Union[Dict, httpx.Headers],
+    ) -> BaseLLMException:
+        """
+        Return appropriate error class for watsonx agent errors.
+
+        Args:
+            error_message: Error message
+            status_code: HTTP status code
+            headers: Response headers
+
+        Returns:
+            WatsonXAIError instance
+        """
+        return WatsonXAIError(
+            status_code=status_code,
+            message=error_message,
+            headers=headers,
+        )
+
+    @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 watsonx agent completion.
+
+        Routes to sync or async completion based on acompletion flag.
+
+        Model format: watsonx_agent/<agent_id>
+
+        Args:
+            model: Model identifier (format: watsonx_agent/<agent_id>)
+            messages: List of messages
+            api_base: API base URL
+            api_key: API key
+            model_response: ModelResponse object to populate
+            logging_obj: Logging object
+            optional_params: Optional parameters
+            litellm_params: LiteLLM parameters
+            timeout: Request timeout
+            acompletion: Async completion flag
+            stream: Streaming flag
+            headers: Request headers
+
+        Returns:
+            ModelResponse object or async coroutine
+        """
+        from litellm.llms.watsonx.agents.handler import watsonx_agent_handler
+
+        if acompletion:
+            return watsonx_agent_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:
+            return watsonx_agent_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,
+            )

litellm/main.py

@@ -2506,10 +2506,10 @@ def completion(  # type: ignore # noqa: PLR0915
 
             # Add GitHub Copilot headers (same as /responses endpoint does)
             if custom_llm_provider == "github_copilot":
+                from litellm.llms.github_copilot.authenticator import Authenticator
                 from litellm.llms.github_copilot.common_utils import (
                     get_copilot_default_headers,
                 )
-                from litellm.llms.github_copilot.authenticator import Authenticator
 
                 copilot_auth = Authenticator()
                 copilot_api_key = copilot_auth.get_api_key()
@@ -3767,6 +3767,25 @@ def completion(  # type: ignore # noqa: PLR0915
                 encoding=_get_encoding(),
                 custom_llm_provider="watsonx",
             )
+        elif custom_llm_provider == "watsonx_agent":
+            # Handle watsonx agent completions
+            # Model format: watsonx_agent/<agent_id>
+            from litellm.llms.watsonx.agents import IBMWatsonXAgentConfig
+
+            response = IBMWatsonXAgentConfig.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,
+            )
         elif custom_llm_provider == "watsonx_text":
             api_key = (
                 api_key

litellm/types/llms/watsonx_agents.py

@@ -0,0 +1,96 @@
+"""
+Type definitions for IBM watsonx.ai Orchestrate Agent API responses.
+
+API Reference: https://developer.watson-orchestrate.ibm.com/apis/orchestrate-agent/chat-with-agents
+"""
+
+from typing import Any, Dict, List, Optional, Union
+
+from typing_extensions import TypedDict
+
+
+class WatsonxAgentMessageContent(TypedDict, total=False):
+    """Content structure for agent messages."""
+
+    response_type: Optional[str]  # e.g., "conversational_search"
+    json_schema: Optional[Dict[str, Any]]
+    ui_schema: Optional[Dict[str, Any]]
+    form_data: Optional[Dict[str, Any]]
+    id: Optional[str]
+    form_operation: Optional[str]
+    sub_type: Optional[str]
+    event_type: Optional[str]
+    dps_payload_id: Optional[str]
+
+
+class WatsonxAgentMessage(TypedDict):
+    """Message structure for watsonx agent requests."""
+
+    role: str
+    content: Union[str, List[WatsonxAgentMessageContent]]
+
+
+class WatsonxAgentAdditionalParameters(TypedDict, total=False):
+    """Additional parameters for agent requests."""
+
+    pass  # Can be extended based on specific needs
+
+
+class WatsonxAgentContext(TypedDict, total=False):
+    """Context dictionary for agent requests."""
+
+    pass  # Optional context information
+
+
+class WatsonxAgentRequestBody(TypedDict):
+    """Request body for watsonx agent chat completions."""
+
+    messages: List[WatsonxAgentMessage]
+    additional_parameters: WatsonxAgentAdditionalParameters
+    context: WatsonxAgentContext
+    stream: bool
+
+
+class WatsonxAgentChoice(TypedDict, total=False):
+    """Choice structure in agent response."""
+
+    index: Optional[int]
+    message: Optional[Dict[str, Any]]
+    finish_reason: Optional[str]
+
+
+class WatsonxAgentResponse(TypedDict):
+    """Response structure from watsonx agent API."""
+
+    id: str
+    object: str
+    created: int
+    model: str
+    choices: List[WatsonxAgentChoice]
+    thread_id: str
+
+
+class WatsonxAgentStreamChunk(TypedDict, total=False):
+    """Streaming response chunk from watsonx agent API."""
+
+    id: str
+    object: str
+    created: int
+    model: str
+    choices: List[WatsonxAgentChoice]
+    thread_id: Optional[str]
+
+
+class WatsonxAgentCredentials(TypedDict):
+    """Credentials for watsonx agent authentication."""
+
+    api_key: str
+    api_base: str
+    token: Optional[str]
+
+
+class WatsonxAgentParams(TypedDict, total=False):
+    """Parameters for watsonx agent API calls."""
+
+    agent_id: str
+    thread_id: Optional[str]

litellm/types/utils.py

@@ -3111,6 +3111,7 @@ class LlmProviders(str, Enum):
     FEATHERLESS_AI = "featherless_ai"
     WATSONX = "watsonx"
     WATSONX_TEXT = "watsonx_text"
+    WATSONX_AGENT = "watsonx_agent"
     TRITON = "triton"
     PREDIBASE = "predibase"
     DATABRICKS = "databricks"

tests/agent_tests/local_only_agent_tests/test_a2a_watsonx_agent.py

@@ -0,0 +1,267 @@
+"""
+Test for WatsonX Agents with A2A Protocol via Completion Bridge.
+
+Tests the A2A SDK-level functions routing WatsonX agent requests through 
+litellm.acompletion using the completion bridge.
+
+Run with:
+    pytest tests/agent_tests/local_only_agent_tests/test_a2a_watsonx_agent.py -v -s
+
+Prerequisites:
+    - WATSONX_API_BASE environment variable set to your WatsonX endpoint
+    - WATSONX_API_KEY environment variable set to your API key
+    - WATSONX_AGENT_ID environment variable set to your agent ID
+"""
+
+import os
+import sys
+from uuid import uuid4
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../.."))
+
+import litellm
+from a2a.types import MessageSendParams, SendMessageRequest, SendStreamingMessageRequest
+
+
+def get_watsonx_config():
+    """Get WatsonX configuration from environment."""
+    api_base = os.environ.get("WATSONX_API_BASE")
+    api_key = os.environ.get("WATSONX_API_KEY")
+    agent_id = os.environ.get("WATSONX_AGENT_ID")
+    
+    if not all([api_base, api_key, agent_id]):
+        pytest.skip("WatsonX credentials not configured. Set WATSONX_API_BASE, WATSONX_API_KEY, and WATSONX_AGENT_ID")
+    
+    return api_base, api_key, agent_id
+
+
+@pytest.mark.asyncio
+async def test_a2a_watsonx_agent_non_streaming():
+    """
+    Test non-streaming A2A request via the completion bridge with WatsonX agent.
+    
+    This test validates that WatsonX agents work with the A2A protocol through
+    the completion bridge, which:
+    1. Receives A2A JSON-RPC request
+    2. Transforms A2A message to OpenAI format
+    3. Routes through litellm.acompletion with watsonx_agent provider
+    4. Transforms response back to A2A format
+    """
+    from litellm.a2a_protocol import asend_message
+
+    api_base, api_key, agent_id = get_watsonx_config()
+    
+    litellm._turn_on_debug()
+
+    send_message_payload = {
+        "message": {
+            "role": "user",
+            "parts": [{"kind": "text", "text": "Hello, introduce yourself in one sentence."}],
+            "messageId": uuid4().hex,
+        }
+    }
+
+    request = SendMessageRequest(
+        id=str(uuid4()),
+        params=MessageSendParams(**send_message_payload),  # type: ignore
+    )
+
+    # Route through completion bridge with watsonx_agent provider
+    response = await asend_message(
+        request=request,
+        api_base=api_base,
+        litellm_params={
+            "custom_llm_provider": "watsonx",
+            "model": f"watsonx_agent/{agent_id}",
+            "api_key": api_key,
+        },
+    )
+
+    # Validate response follows A2A SendMessageResponse format
+    assert response.jsonrpc == "2.0"
+    assert response.id is not None
+    assert response.result is not None
+    assert "message" in response.result
+
+    message = response.result["message"]
+    assert "role" in message
+    assert message["role"] == "agent"
+    assert "parts" in message
+    assert len(message["parts"]) > 0
+    assert message["parts"][0]["kind"] == "text"
+    assert len(message["parts"][0]["text"]) > 0
+
+    print(f"\nWatsonX A2A Response: {response.model_dump(mode='json', exclude_none=True)}")
+    print(f"Agent said: {message['parts'][0]['text']}")
+
+
+@pytest.mark.asyncio
+async def test_a2a_watsonx_agent_streaming():
+    """
+    Test streaming A2A request via the completion bridge with WatsonX agent.
+
+    Validates proper A2A streaming format with events:
+    1. Task event (kind: "task") - Initial task with status "submitted"
+    2. Status update (kind: "status-update") - Status "working"
+    3. Artifact update (kind: "artifact-update") - Content delivery
+    4. Status update (kind: "status-update") - Final "completed" status
+    """
+    from litellm.a2a_protocol import asend_message_streaming
+
+    api_base, api_key, agent_id = get_watsonx_config()
+    
+    litellm._turn_on_debug()
+
+    send_message_payload = {
+        "message": {
+            "role": "user",
+            "parts": [{"kind": "text", "text": "List three benefits of AI in one sentence each."}],
+            "messageId": uuid4().hex,
+        }
+    }
+
+    request = SendStreamingMessageRequest(
+        id=str(uuid4()),
+        params=MessageSendParams(**send_message_payload),  # type: ignore
+    )
+
+    chunks = []
+    async for chunk in asend_message_streaming(
+        request=request,
+        api_base=api_base,
+        litellm_params={
+            "custom_llm_provider": "watsonx",
+            "model": f"watsonx_agent/{agent_id}",
+            "api_key": api_key,
+        },
+    ):
+        chunks.append(chunk)
+        print(f"Chunk: {chunk}")
+
+    # Validate we received proper A2A streaming events
+    assert len(chunks) >= 4, f"Expected at least 4 chunks (task, working, artifact, completed), got {len(chunks)}"
+
+    # Validate chunk structure follows A2A spec
+    for chunk in chunks:
+        assert "jsonrpc" in chunk
+        assert chunk["jsonrpc"] == "2.0"
+        assert "id" in chunk
+        assert "result" in chunk
+
+    # Validate first chunk is task event
+    task_chunk = chunks[0]
+    assert task_chunk["result"]["kind"] == "task", "First chunk should be task event"
+    assert task_chunk["result"]["status"]["state"] == "submitted"
+    assert "contextId" in task_chunk["result"]
+    assert "id" in task_chunk["result"]  # task id
+    assert "history" in task_chunk["result"]
+
+    # Validate second chunk is working status update
+    working_chunk = chunks[1]
+    assert working_chunk["result"]["kind"] == "status-update", "Second chunk should be status-update"
+    assert working_chunk["result"]["status"]["state"] == "working"
+    assert "taskId" in working_chunk["result"]
+    assert "contextId" in working_chunk["result"]
+    assert working_chunk["result"]["final"] is False
+
+    # Validate artifact update chunk
+    artifact_chunk = chunks[2]
+    assert artifact_chunk["result"]["kind"] == "artifact-update", "Third chunk should be artifact-update"
+    assert "artifact" in artifact_chunk["result"]
+    assert "artifactId" in artifact_chunk["result"]["artifact"]
+    assert "parts" in artifact_chunk["result"]["artifact"]
+    assert len(artifact_chunk["result"]["artifact"]["parts"]) > 0
+    assert artifact_chunk["result"]["artifact"]["parts"][0]["kind"] == "text"
+
+    # Validate final chunk is completed status update
+    final_chunk = chunks[-1]
+    assert final_chunk["result"]["kind"] == "status-update", "Last chunk should be status-update"
+    assert final_chunk["result"]["status"]["state"] == "completed"
+    assert final_chunk["result"]["final"] is True
+
+    print(f"\nReceived {len(chunks)} chunks with proper A2A streaming format")
+    print(f"Agent response: {artifact_chunk['result']['artifact']['parts'][0]['text']}")
+
+
+@pytest.mark.asyncio
+async def test_a2a_watsonx_agent_with_thread_continuity():
+    """
+    Test WatsonX agent with thread continuity through A2A protocol.
+    
+    WatsonX agents support thread-based conversation continuity. This test
+    validates that thread_id is properly passed through the A2A bridge.
+    """
+    from litellm.a2a_protocol import asend_message
+
+    api_base, api_key, agent_id = get_watsonx_config()
+    
+    litellm._turn_on_debug()
+
+    # First message - creates a new thread
+    send_message_payload_1 = {
+        "message": {
+            "role": "user",
+            "parts": [{"kind": "text", "text": "Remember this: my favorite color is blue."}],
+            "messageId": uuid4().hex,
+        }
+    }
+
+    request1 = SendMessageRequest(
+        id=str(uuid4()),
+        params=MessageSendParams(**send_message_payload_1),  # type: ignore
+    )
+
+    response1 = await asend_message(
+        request=request1,
+        api_base=api_base,
+        litellm_params={
+            "custom_llm_provider": "watsonx",
+            "model": f"watsonx_agent/{agent_id}",
+            "api_key": api_key,
+        },
+    )
+
+    print(f"\nFirst message response: {response1.model_dump(mode='json', exclude_none=True)}")
+
+    # Extract thread_id from hidden params (if available)
+    thread_id = response1._hidden_params.get("thread_id") if hasattr(response1, "_hidden_params") else None
+    
+    if thread_id:
+        print(f"Thread ID: {thread_id}")
+        
+        # Second message - continue conversation with same thread
+        send_message_payload_2 = {
+            "message": {
+                "role": "user",
+                "parts": [{"kind": "text", "text": "What is my favorite color?"}],
+                "messageId": uuid4().hex,
+            }
+        }
+
+        request2 = SendMessageRequest(
+            id=str(uuid4()),
+            params=MessageSendParams(**send_message_payload_2),  # type: ignore
+        )
+
+        response2 = await asend_message(
+            request=request2,
+            api_base=api_base,
+            litellm_params={
+                "custom_llm_provider": "watsonx",
+                "model": f"watsonx_agent/{agent_id}",
+                "api_key": api_key,
+                "thread_id": thread_id,  # Continue same conversation
+            },
+        )
+
+        print(f"\nSecond message response: {response2.model_dump(mode='json', exclude_none=True)}")
+        
+        # The agent should remember the color from the first message
+        response_text = response2.result["message"]["parts"][0]["text"].lower()
+        print(f"Agent's answer: {response_text}")
+        # Note: Asserting the content might be flaky depending on agent behavior
+        # assert "blue" in response_text, "Agent should remember the favorite color"
+    else:
+        print("Thread ID not available in response - skipping continuity test")

tests/test_litellm/llms/watsonx/agents/__init__.py

@@ -0,0 +1,3 @@
+"""
+Tests for IBM watsonx.ai Orchestrate Agent support.
+"""

tests/test_litellm/llms/watsonx/agents/test_watsonx_agents_transformation.py

@@ -0,0 +1,378 @@
+"""
+Tests for IBM watsonx.ai Orchestrate Agent transformation.
+"""
+
+import json
+import os
+import sys
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+from httpx import Response
+
+sys.path.insert(
+    0, os.path.abspath("../../../../..")
+)  # Adds the parent directory to the system path
+
+from litellm.llms.watsonx.agents.transformation import IBMWatsonXAgentConfig
+from litellm.types.utils import Message, ModelResponse
+
+
+class TestIBMWatsonXAgentConfig:
+    """Test suite for IBMWatsonXAgentConfig methods"""
+
+    @pytest.fixture
+    def config(self):
+        """Create a test instance of IBMWatsonXAgentConfig"""
+        return IBMWatsonXAgentConfig()
+
+    @pytest.fixture
+    def sample_messages(self):
+        """Sample messages for testing"""
+        return [
+            {"role": "user", "content": "Hello, how can you help me?"},
+            {"role": "assistant", "content": "I can help with various tasks."},
+            {"role": "user", "content": "What is the weather like?"},
+        ]
+
+    @pytest.fixture
+    def sample_agent_response(self):
+        """Sample agent API response"""
+        return {
+            "id": "chatcmpl-123",
+            "object": "chat.completion",
+            "created": 1677652288,
+            "model": "watsonx_agent/abc123",
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": "The weather is sunny today.",
+                    },
+                    "finish_reason": "stop",
+                }
+            ],
+            "thread_id": "thread-xyz789",
+        }
+
+    def test_get_supported_openai_params(self, config):
+        """Test getting supported OpenAI parameters"""
+        model = "watsonx_agent/test123"
+        params = config.get_supported_openai_params(model)
+
+        assert "stream" in params
+        assert "messages" in params
+
+    def test_get_agent_id_valid(self, config):
+        """Test extracting valid agent_id from model string"""
+        model = "watsonx_agent/abc123"
+        agent_id = config._get_agent_id(model)
+
+        assert agent_id == "abc123"
+
+    def test_get_agent_id_complex(self, config):
+        """Test extracting agent_id with complex format"""
+        model = "watsonx_agent/agent-id-with-dashes"
+        agent_id = config._get_agent_id(model)
+
+        assert agent_id == "agent-id-with-dashes"
+
+    def test_get_agent_id_invalid_format(self, config):
+        """Test extracting agent_id with invalid format raises error"""
+        invalid_models = [
+            "invalid_model",
+            "watsonx_agent",
+            "",
+        ]
+
+        for invalid_model in invalid_models:
+            with pytest.raises(ValueError, match="Invalid model format"):
+                config._get_agent_id(invalid_model)
+
+    @patch.object(IBMWatsonXAgentConfig, "_get_base_url")
+    def test_get_complete_url(self, mock_get_base_url, config):
+        """Test building complete URL"""
+        mock_get_base_url.return_value = "https://api.example.com"
+
+        model = "watsonx_agent/abc123"
+        url = config.get_complete_url(
+            api_base=None,
+            api_key="test-key",
+            model=model,
+            optional_params={},
+            litellm_params={},
+            stream=False,
+        )
+
+        expected_url = "https://api.example.com/api/v1/orchestrate/abc123/chat/completions"
+        assert url == expected_url
+
+    @patch.object(IBMWatsonXAgentConfig, "_get_base_url")
+    def test_get_complete_url_with_trailing_slash(self, mock_get_base_url, config):
+        """Test building complete URL with trailing slash in base URL"""
+        mock_get_base_url.return_value = "https://api.example.com/"
+
+        model = "watsonx_agent/abc123"
+        url = config.get_complete_url(
+            api_base=None,
+            api_key="test-key",
+            model=model,
+            optional_params={},
+            litellm_params={},
+            stream=False,
+        )
+
+        expected_url = "https://api.example.com/api/v1/orchestrate/abc123/chat/completions"
+        assert url == expected_url
+
+    def test_transform_messages(self, config, sample_messages):
+        """Test transforming OpenAI messages to agent format"""
+        transformed = config._transform_messages(sample_messages)
+
+        assert len(transformed) == 3
+        assert transformed[0]["role"] == "user"
+        assert transformed[0]["content"] == "Hello, how can you help me?"
+        assert transformed[2]["role"] == "user"
+        assert transformed[2]["content"] == "What is the weather like?"
+
+    def test_transform_messages_with_list_content(self, config):
+        """Test transforming messages with list content"""
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "Hello"},
+                    {"type": "text", "text": " world"},
+                ],
+            }
+        ]
+
+        with patch(
+            "litellm.llms.watsonx.agents.transformation.convert_content_list_to_str"
+        ) as mock_convert:
+            mock_convert.return_value = "Hello world"
+            transformed = config._transform_messages(messages)
+
+            assert len(transformed) == 1
+            assert transformed[0]["content"] == "Hello world"
+
+    def test_transform_request(self, config, sample_messages):
+        """Test transforming complete request"""
+        model = "watsonx_agent/abc123"
+        optional_params = {
+            "stream": True,
+            "additional_parameters": {"key": "value"},
+            "context": {"user_id": "123"},
+        }
+
+        request_data = config.transform_request(
+            model=model,
+            messages=sample_messages,
+            optional_params=optional_params,
+            litellm_params={},
+            headers={},
+        )
+
+        assert "messages" in request_data
+        assert len(request_data["messages"]) == 3
+        assert request_data["stream"] is True
+        assert request_data["additional_parameters"] == {"key": "value"}
+        assert request_data["context"] == {"user_id": "123"}
+
+    def test_transform_request_defaults(self, config, sample_messages):
+        """Test transforming request with default values"""
+        model = "watsonx_agent/abc123"
+        optional_params = {}
+
+        request_data = config.transform_request(
+            model=model,
+            messages=sample_messages,
+            optional_params=optional_params,
+            litellm_params={},
+            headers={},
+        )
+
+        assert "messages" in request_data
+        assert request_data["stream"] is True
+        assert request_data["additional_parameters"] == {}
+        assert request_data["context"] == {}
+
+    def test_transform_agent_choice_to_litellm(self, config):
+        """Test transforming agent choice to LiteLLM format"""
+        agent_choice = {
+            "index": 0,
+            "message": {"role": "assistant", "content": "Hello!"},
+            "finish_reason": "stop",
+        }
+
+        litellm_choice = config._transform_agent_choice_to_litellm(agent_choice, 0)
+
+        assert litellm_choice.index == 0
+        assert litellm_choice.message.role == "assistant"
+        assert litellm_choice.message.content == "Hello!"
+        assert litellm_choice.finish_reason == "stop"
+
+    def test_transform_agent_choice_with_dict_content(self, config):
+        """Test transforming agent choice with dict content"""
+        agent_choice = {
+            "index": 0,
+            "message": {"role": "assistant", "content": {"key": "value"}},
+            "finish_reason": "stop",
+        }
+
+        litellm_choice = config._transform_agent_choice_to_litellm(agent_choice, 0)
+
+        assert litellm_choice.message.content == json.dumps({"key": "value"})
+
+    def test_transform_agent_choice_with_list_content(self, config):
+        """Test transforming agent choice with list content"""
+        agent_choice = {
+            "index": 0,
+            "message": {"role": "assistant", "content": ["item1", "item2"]},
+            "finish_reason": "stop",
+        }
+
+        litellm_choice = config._transform_agent_choice_to_litellm(agent_choice, 0)
+
+        assert litellm_choice.message.content == json.dumps(["item1", "item2"])
+
+    def test_transform_agent_choice_missing_finish_reason(self, config):
+        """Test transforming agent choice without finish_reason"""
+        agent_choice = {
+            "index": 0,
+            "message": {"role": "assistant", "content": "Hello!"},
+        }
+
+        litellm_choice = config._transform_agent_choice_to_litellm(agent_choice, 0)
+
+        assert litellm_choice.finish_reason == "stop"
+
+    def test_transform_response(self, config, sample_agent_response):
+        """Test transforming complete response"""
+        # Create mock response
+        mock_response = Mock(spec=Response)
+        mock_response.json.return_value = sample_agent_response
+        mock_response.status_code = 200
+
+        model_response = ModelResponse()
+
+        result = config.transform_response(
+            model="watsonx_agent/abc123",
+            raw_response=mock_response,
+            model_response=model_response,
+            logging_obj=None,
+            request_data={},
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            encoding=None,
+            api_key=None,
+        )
+
+        assert result.id == "chatcmpl-123"
+        assert result.created == 1677652288
+        assert result.model == "watsonx_agent/abc123"
+        assert len(result.choices) == 1
+        assert result.choices[0].message.content == "The weather is sunny today."
+        assert result._hidden_params is not None
+        assert result._hidden_params.get("thread_id") == "thread-xyz789"
+
+    def test_transform_response_multiple_choices(self, config):
+        """Test transforming response with multiple choices"""
+        agent_response = {
+            "id": "chatcmpl-456",
+            "object": "chat.completion",
+            "created": 1677652288,
+            "model": "watsonx_agent/abc123",
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {"role": "assistant", "content": "Response 1"},
+                    "finish_reason": "stop",
+                },
+                {
+                    "index": 1,
+                    "message": {"role": "assistant", "content": "Response 2"},
+                    "finish_reason": "stop",
+                },
+            ],
+            "thread_id": "thread-xyz789",
+        }
+
+        mock_response = Mock(spec=Response)
+        mock_response.json.return_value = agent_response
+        mock_response.status_code = 200
+
+        model_response = ModelResponse()
+
+        result = config.transform_response(
+            model="watsonx_agent/abc123",
+            raw_response=mock_response,
+            model_response=model_response,
+            logging_obj=None,
+            request_data={},
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            encoding=None,
+            api_key=None,
+        )
+
+        assert len(result.choices) == 2
+        assert result.choices[0].message.content == "Response 1"
+        assert result.choices[1].message.content == "Response 2"
+
+    def test_map_openai_params(self, config):
+        """Test mapping OpenAI parameters"""
+        non_default_params = {
+            "thread_id": "thread-123",
+            "additional_parameters": {"key": "value"},
+            "context": {"user": "test"},
+            "unsupported_param": "should_be_ignored",
+        }
+        optional_params = {}
+
+        result = config.map_openai_params(
+            non_default_params=non_default_params,
+            optional_params=optional_params,
+            model="watsonx_agent/abc123",
+            drop_params=False,
+        )
+
+        assert result["thread_id"] == "thread-123"
+        assert result["additional_parameters"] == {"key": "value"}
+        assert result["context"] == {"user": "test"}
+        # unsupported_param should remain in non_default_params
+        assert "unsupported_param" in non_default_params
+
+    @patch.object(IBMWatsonXAgentConfig, "validate_environment")
+    def test_validate_environment_called(self, mock_validate, config):
+        """Test that validate_environment uses parent class method"""
+        mock_validate.return_value = {"Authorization": "Bearer token"}
+
+        headers = config.validate_environment(
+            headers={},
+            model="watsonx_agent/abc123",
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            api_key="test-key",
+            api_base="https://api.example.com",
+        )
+
+        mock_validate.assert_called_once()
+
+    def test_get_error_class(self, config):
+        """Test getting error class"""
+        from litellm.llms.watsonx.common_utils import WatsonXAIError
+
+        error = config.get_error_class(
+            error_message="Test error",
+            status_code=400,
+            headers={},
+        )
+
+        assert isinstance(error, WatsonXAIError)
+        assert error.status_code == 400
+        assert "Test error" in error.message