feat(a2a): watsonx Orchestrate agent provider

[Sameerlite]

Summary

• Add watsonx_orchestrate A2A provider (auth, submit run, poll, SSE + fallback streaming).
• Register provider in A2AProviderConfigManager.
• Add dashboard agent type in agent_create_fields.json.
• Unit tests for WXO request/response transformation helpers.
  Fix LIT-1967

Test plan

• pytest tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py -v
• Create agent in UI with watsonx Orchestrate type and invoke via /a2a/{agent}/message/send

---

Note

Medium Risk
Introduces new outbound auth (API keys, bearer tokens) and long-running poll/stream HTTP to customer CP4D/IBM endpoints; failures are mostly isolated to the new provider but misconfiguration can leak credentials into logs if existing patterns do.

Overview
Adds a watsonx_orchestrate A2A provider so LiteLLM can drive IBM watsonx Orchestrate agents through the existing agent/A2A path.

Provider wiring: A2AProviderConfigManager resolves custom_llm_provider=watsonx_orchestrate to a new config/handler stack that translates A2A send/stream requests into WXO REST calls (POST /v1/orchestrate/runs, poll GET .../runs/{id}, optional POST .../runs/stream), then maps results back to A2A JSON-RPC (non-streaming message responses and streaming task/status/artifact events, including fake streaming when the upstream response is already complete).

Auth & params: Supports CP4D (username + API key → cluster authorize) and IBM Cloud IAM (API key only), with in-memory bearer token caching. Required litellm_params: cp4d_host, instance_id, wxo_agent_id, api_key; optional auth_mode, thread_id, username (enforced at runtime for CP4D).

Proxy UI: agent_create_fields.json gains a watsonx Orchestrate agent type with credential fields and custom_llm_provider: watsonx_orchestrate.

Tests: New unit tests cover transformations, token TTL, SSE accumulation, streaming poll/failure paths, config manager registration, and dashboard field defaults.

^{Reviewed by Cursor Bugbot for commit 900926a. Bugbot is set up for automated code reviews on this repo. Configure here.}

[greptile-apps]

Greptile Summary

This PR adds a watsonx_orchestrate A2A provider that bridges LiteLLM's A2A path to IBM watsonx Orchestrate's REST runs API, including CP4D and IBM Cloud IAM auth with in-memory token caching, poll/SSE streaming, and a fake-streaming fallback.

• New provider module (handler.py, config.py, transformation.py) wires WXO auth, run submission, polling, and SSE accumulation through litellm's existing get_async_httpx_client infrastructure; the A2AProviderConfigManager registers the provider with a lazy import.
• The dashboard agent_create_fields.json gains a watsonx_orchestrate entry with credential fields matching the handler's required litellm_params.
• Unit tests cover transformations, token TTL/caching, SSE accumulation, poll-timeout, streaming fallback, and the config manager registration — all mocked, no real network calls.

Confidence Score: 5/5

The PR is safe to merge; the new provider is self-contained, uses litellm's shared HTTP infrastructure, and does not touch any existing request paths.

All three major issues flagged in the previous review round have been corrected: the streaming path now uses get_async_httpx_client instead of a raw httpx.AsyncClient(verify=False), the try/except httpx.TransportError block covers only the initial POST, and the CP4D expiration field is now correctly treated as an absolute Unix timestamp. The remaining findings are style-level redundancies that do not affect correctness.

No files require special attention; handler.py carries the most complexity but the logic is well-tested and the auth paths are correct.

Important Files Changed

Filename	Overview
litellm/a2a_protocol/providers/watsonx_orchestrate/handler.py	Core handler with auth, polling, SSE accumulation, and streaming fallback; previously flagged issues have been addressed; explicit response.raise_for_status() after client.post() is redundant since AsyncHTTPHandler.post already calls it internally
litellm/a2a_protocol/providers/watsonx_orchestrate/transformation.py	Transformation helpers for A2A↔WXO mapping and fake streaming; well-structured with multiple fallback paths for text extraction
litellm/a2a_protocol/providers/config_manager.py	Registers watsonx_orchestrate provider with lazy import; follows the same pattern as BedrockAgentCore
litellm/proxy/public_endpoints/agent_create_fields.json	Adds watsonx Orchestrate dashboard agent type; field definitions match the handler's _extract_litellm_params requirements
tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py	Good mock-only test coverage for transformation helpers, token TTL, cache behavior, SSE accumulation, and streaming fallback/poll paths; assertions are correct

_{Reviews (11): Last reviewed commit: "fix(a2a/wxo): evict expired token cache ..." | Re-trigger Greptile}

[codecov]

Codecov Report

❌ Patch coverage is 88.84758% with 30 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
..._protocol/providers/watsonx_orchestrate/handler.py	91.66%	14 Missing ⚠️
...a_protocol/providers/watsonx_orchestrate/config.py	40.00%	9 Missing ⚠️
...ol/providers/watsonx_orchestrate/transformation.py	91.56%	7 Missing ⚠️

📢 Thoughts on this report? Let us know!

[veria-ai]

PR overview

All previously flagged issues have been addressed. No open security concerns remain on this pull request.

Security review

No open security issues remain on this pull request.

Fixed/addressed: 1 · PR risk: 0/10

[Sameerlite]

@greptileai

[Sameerlite]

@greptileai Addressed P1: CP4D expiration is now interpreted as an absolute Unix timestamp (expiration - time.time()), not added to time.monotonic(). IBM Cloud IAM expires_in path unchanged. Added unit test test_cp4d_token_ttl_from_absolute_expiration.

[Sameerlite]

@greptileai

[Sameerlite]

bugbot run

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 3 potential issues.

Autofix Details

Bugbot Autofix prepared fixes for all 3 issues found in the latest run.

• ✅ Fixed: Text extraction elif matches non-text kind parts
  • A2A text extraction now only includes parts whose kind is text, missing, or empty, with regression coverage for non-text parts carrying text fields.
• ✅ Fixed: SSE parser crashes on non-dict JSON events
  • WXO result text extraction now returns an empty string for non-dict JSON values, so SSE accumulation ignores keepalive-style events without aborting.
• ✅ Fixed: Token cache minimum can serve expired tokens
  • Token cache expiry now floors at zero after the safety buffer, preventing short-lived tokens from being served past their actual expiration.

_{You can send follow-ups to the cloud agent here.}

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution.
2 out of 3 committers have signed the CLA.

✅ Sameerlite
✅ mateo-berri
❌ cursoragent
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[Sameerlite]

bugbot run

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Broad exception catch silently swallows streaming errors causing double requests
  • Restricted the WXO streaming fallback to transport-level failures so HTTP/status, JSON parsing, and other unexpected errors propagate without a second request.

Preview (3fcbc4ceac)

diff --git a/litellm/a2a_protocol/providers/config_manager.py b/litellm/a2a_protocol/providers/config_manager.py
--- a/litellm/a2a_protocol/providers/config_manager.py
+++ b/litellm/a2a_protocol/providers/config_manager.py
@@ -48,4 +48,11 @@
 
             return BedrockAgentCoreA2AConfig()
 
+        if custom_llm_provider == "watsonx_orchestrate":
+            from litellm.a2a_protocol.providers.watsonx_orchestrate.config import (
+                WatsonxOrchestrateA2AConfig,
+            )
+
+            return WatsonxOrchestrateA2AConfig()
+
         return None

diff --git a/litellm/a2a_protocol/providers/watsonx_orchestrate/__init__.py b/litellm/a2a_protocol/providers/watsonx_orchestrate/__init__.py
new file mode 100644
--- /dev/null
+++ b/litellm/a2a_protocol/providers/watsonx_orchestrate/__init__.py
@@ -1,0 +1,3 @@
+"""
+IBM watsonx Orchestrate (WXO) A2A provider.
+"""

diff --git a/litellm/a2a_protocol/providers/watsonx_orchestrate/config.py b/litellm/a2a_protocol/providers/watsonx_orchestrate/config.py
new file mode 100644
--- /dev/null
+++ b/litellm/a2a_protocol/providers/watsonx_orchestrate/config.py
@@ -1,0 +1,55 @@
+"""
+A2A provider configuration for IBM watsonx Orchestrate (WXO).
+"""
+
+from typing import Any, AsyncIterator, Dict, Optional
+
+from litellm.a2a_protocol.providers.base import BaseA2AProviderConfig
+from litellm.a2a_protocol.providers.watsonx_orchestrate.handler import (
+    WatsonxOrchestrateHandler,
+)
+
+
+class WatsonxOrchestrateA2AConfig(BaseA2AProviderConfig):
+    """A2A bridge for IBM watsonx Orchestrate (REST runs API + poll/SSE)."""
+
+    async def handle_non_streaming(
+        self,
+        request_id: str,
+        params: Dict[str, Any],
+        api_base: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """Handle a non-streaming A2A request via WXO runs API."""
+        litellm_params = kwargs.get("litellm_params")
+        if not litellm_params:
+            raise ValueError(
+                "litellm_params is required for WatsonxOrchestrateA2AConfig "
+                "(must contain cp4d_host, instance_id, wxo_agent_id, api_key)"
+            )
+        return await WatsonxOrchestrateHandler.handle_non_streaming(
+            request_id=request_id,
+            params=params,
+            litellm_params=litellm_params,
+        )
+
+    async def handle_streaming(
+        self,
+        request_id: str,
+        params: Dict[str, Any],
+        api_base: Optional[str] = None,
+        **kwargs: Any,
+    ) -> AsyncIterator[Dict[str, Any]]:
+        """Handle a streaming A2A request via WXO streaming runs API."""
+        litellm_params = kwargs.get("litellm_params")
+        if not litellm_params:
+            raise ValueError(
+                "litellm_params is required for WatsonxOrchestrateA2AConfig "
+                "(must contain cp4d_host, instance_id, wxo_agent_id, api_key)"
+            )
+        async for chunk in WatsonxOrchestrateHandler.handle_streaming(
+            request_id=request_id,
+            params=params,
+            litellm_params=litellm_params,
+        ):
+            yield chunk

diff --git a/litellm/a2a_protocol/providers/watsonx_orchestrate/handler.py b/litellm/a2a_protocol/providers/watsonx_orchestrate/handler.py
new file mode 100644
--- /dev/null
+++ b/litellm/a2a_protocol/providers/watsonx_orchestrate/handler.py
@@ -1,0 +1,360 @@
+"""
+Handler for IBM watsonx Orchestrate (WXO) agent provider.
+"""
+
+import asyncio
+import hashlib
+import json
+import time
+from typing import Any, AsyncIterator, Dict, Optional, Tuple, cast
+
+import httpx
+
+from litellm._logging import verbose_logger
+from litellm.a2a_protocol.providers.watsonx_orchestrate.transformation import (
+    WatsonxOrchestrateTransformation,
+)
+from litellm.llms.custom_httpx.http_handler import (
+    AsyncHTTPHandler,
+    get_async_httpx_client,
+)
+from litellm.types.llms.custom_http import httpxSpecialProvider
+
+_IBM_CLOUD_IAM_URL = "https://iam.cloud.ibm.com/identity/token"
+_POLL_INTERVAL_S = 2.0
+_MAX_POLL_ATTEMPTS = 90
+_TOKEN_CACHE_TTL_BUFFER_S = 60
+_token_cache: Dict[str, Tuple[str, float]] = {}
+
+
+class WatsonxOrchestrateHandler:
+    @staticmethod
+    def _http_client(timeout: float = 90.0) -> AsyncHTTPHandler:
+        return get_async_httpx_client(
+            llm_provider=cast(Any, httpxSpecialProvider.A2AProvider),
+            params={"timeout": timeout},
+        )
+
+    @staticmethod
+    def _token_cache_key(
+        auth_mode: str,
+        cp4d_host: str,
+        api_key: str,
+        username: Optional[str],
+    ) -> str:
+        material = f"{auth_mode}:{cp4d_host}:{username or ''}:{api_key}"
+        return hashlib.sha256(material.encode()).hexdigest()
+
+    @staticmethod
+    def _cp4d_token_ttl_seconds(
+        expiration: Any, now_wall: Optional[float] = None
+    ) -> int:
+        # CP4D returns expiration as absolute Unix epoch seconds, not a duration.
+        expires_at = int(expiration)
+        wall = now_wall if now_wall is not None else time.time()
+        return max(expires_at - int(wall), 0)
+
+    @staticmethod
+    async def _get_bearer_token(
+        cp4d_host: str,
+        auth_mode: str,
+        api_key: str,
+        username: Optional[str] = None,
+        client: Optional[AsyncHTTPHandler] = None,
+    ) -> str:
+        cache_key = WatsonxOrchestrateHandler._token_cache_key(
+            auth_mode, cp4d_host, api_key, username
+        )
+        now = time.monotonic()
+        cached = _token_cache.get(cache_key)
+        if cached and cached[1] > now:
+            return cached[0]
+
+        if client is None:
+            client = WatsonxOrchestrateHandler._http_client(timeout=30.0)
+
+        if auth_mode == "ibm_cloud":
+            response = await client.post(
+                _IBM_CLOUD_IAM_URL,
+                data={
+                    "grant_type": "urn:ibm:params:oauth:grant-type:apikey",
+                    "apikey": api_key,
+                },
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+            )
+            response.raise_for_status()
+            payload = response.json()
+            token = str(payload["access_token"])
+            ttl_s = int(payload.get("expires_in", 3600))
+        else:
+            if not username:
+                raise ValueError(
+                    "'username' is required in litellm_params when auth_mode='cp4d'"
+                )
+            token_url = f"{cp4d_host.rstrip('/')}/icp4d-api/v1/authorize"
+            response = await client.post(
+                token_url,
+                json={"username": username, "api_key": api_key},
+                headers={"Content-Type": "application/json"},
+            )
+            response.raise_for_status()
+            payload = response.json()
+            token = str(payload["token"])
+            expiration = payload.get("expiration")
+            if expiration is None:
+                ttl_s = 3600
+            else:
+                ttl_s = WatsonxOrchestrateHandler._cp4d_token_ttl_seconds(expiration)
+
+        expires_at = now + max(ttl_s - _TOKEN_CACHE_TTL_BUFFER_S, 0)
+        _token_cache[cache_key] = (token, expires_at)
+        return token
+
+    @staticmethod
+    async def _poll_run(
+        base_url: str,
+        run_id: str,
+        auth_headers: Dict[str, str],
+        client: AsyncHTTPHandler,
+        max_attempts: int = _MAX_POLL_ATTEMPTS,
+        interval_s: float = _POLL_INTERVAL_S,
+    ) -> Dict[str, Any]:
+        url = f"{base_url}/v1/orchestrate/runs/{run_id}"
+
+        for attempt in range(max_attempts):
+            await asyncio.sleep(interval_s)
+            response = await client.get(url, headers=auth_headers)
+            response.raise_for_status()
+            result: Dict[str, Any] = response.json()
+            status = result.get("status", "")
+            verbose_logger.debug(
+                f"WXO: Poll {attempt + 1}/{max_attempts} run='{run_id}' status='{status}'"
+            )
+            if status in WatsonxOrchestrateTransformation.TERMINAL_STATES:
+                return result
+
+        raise TimeoutError(
+            f"WXO run '{run_id}' did not reach a terminal state after "
+            f"{max_attempts * interval_s:.0f}s"
+        )
+
+    @staticmethod
+    async def _accumulate_wxo_sse_text(response: Any) -> str:
+        accumulated_text = ""
+        async for line in response.aiter_lines():
+            if not line.startswith("data:"):
+                continue
+            data_str = line[5:].strip()
+            if not data_str or data_str == "[DONE]":
+                continue
+            try:
+                event = json.loads(data_str)
+            except json.JSONDecodeError:
+                continue
+            chunk_text = WatsonxOrchestrateTransformation.extract_text_from_wxo_result(
+                event
+            )
+            if chunk_text:
+                accumulated_text += chunk_text
+        return accumulated_text
+
+    @staticmethod
+    def _extract_litellm_params(litellm_params: Dict[str, Any]) -> tuple:
+        cp4d_host = litellm_params.get("cp4d_host") or ""
+        instance_id = litellm_params.get("instance_id") or ""
+        wxo_agent_id = litellm_params.get("wxo_agent_id") or ""
+        api_key = litellm_params.get("api_key") or ""
+        username = litellm_params.get("username") or None
+        auth_mode = litellm_params.get("auth_mode") or "cp4d"
+        thread_id = litellm_params.get("thread_id") or None
+
+        if not cp4d_host:
+            raise ValueError("'cp4d_host' is required in litellm_params for WXO agents")
+        if not instance_id:
+            raise ValueError(
+                "'instance_id' is required in litellm_params for WXO agents"
+            )
+        if not wxo_agent_id:
+            raise ValueError(
+                "'wxo_agent_id' is required in litellm_params for WXO agents"
+            )
+        if not api_key:
+            raise ValueError("'api_key' is required in litellm_params for WXO agents")
+
+        return (
+            cp4d_host,
+            instance_id,
+            wxo_agent_id,
+            api_key,
+            username,
+            auth_mode,
+            thread_id,
+        )
+
+    @staticmethod
+    async def handle_non_streaming(
+        request_id: str,
+        params: Dict[str, Any],
+        litellm_params: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        (
+            cp4d_host,
+            instance_id,
+            wxo_agent_id,
+            api_key,
+            username,
+            auth_mode,
+            thread_id,
+        ) = WatsonxOrchestrateHandler._extract_litellm_params(litellm_params)
+
+        client = WatsonxOrchestrateHandler._http_client(timeout=90.0)
+        token = await WatsonxOrchestrateHandler._get_bearer_token(
+            cp4d_host=cp4d_host,
+            auth_mode=auth_mode,
+            api_key=api_key,
+            username=username,
+            client=client,
+        )
+        base_url = WatsonxOrchestrateTransformation.get_api_base_url(
+            cp4d_host, instance_id
+        )
+        auth_headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        }
+
+        text = WatsonxOrchestrateTransformation.extract_text_from_a2a_params(params)
+        body = WatsonxOrchestrateTransformation.build_wxo_run_body(
+            wxo_agent_id=wxo_agent_id, text=text, thread_id=thread_id
+        )
+
+        run_response = await client.post(
+            f"{base_url}/v1/orchestrate/runs",
+            json=body,
+            headers=auth_headers,
+        )
+        run_response.raise_for_status()
+        run_data: Dict[str, Any] = run_response.json()
+
+        status = run_data.get("status", "")
+        if status not in WatsonxOrchestrateTransformation.TERMINAL_STATES:
+            run_id = run_data.get("run_id") or run_data.get("id") or ""
+            if not run_id:
+                raise ValueError(f"WXO: No run_id in response: {run_data}")
+            run_data = await WatsonxOrchestrateHandler._poll_run(
+                base_url=base_url,
+                run_id=run_id,
+                auth_headers=auth_headers,
+                client=client,
+            )
+            status = run_data.get("status", "")
+
+        if status not in WatsonxOrchestrateTransformation.SUCCESS_STATES:
+            raise RuntimeError(
+                f"WXO run ended with non-success status '{status}': {run_data}"
+            )
+
+        response_text = WatsonxOrchestrateTransformation.extract_text_from_wxo_result(
+            run_data
+        )
+        return WatsonxOrchestrateTransformation.build_a2a_message_response(
+            request_id=request_id, text=response_text
+        )
+
+    @staticmethod
+    async def handle_streaming(
+        request_id: str,
+        params: Dict[str, Any],
+        litellm_params: Dict[str, Any],
+        chunk_size: int = 50,
+        delay_ms: int = 10,
+    ) -> AsyncIterator[Dict[str, Any]]:
+        (
+            cp4d_host,
+            instance_id,
+            wxo_agent_id,
+            api_key,
+            username,
+            auth_mode,
+            thread_id,
+        ) = WatsonxOrchestrateHandler._extract_litellm_params(litellm_params)
+
+        client = WatsonxOrchestrateHandler._http_client(timeout=120.0)
+        token = await WatsonxOrchestrateHandler._get_bearer_token(
+            cp4d_host=cp4d_host,
+            auth_mode=auth_mode,
+            api_key=api_key,
+            username=username,
+            client=client,
+        )
+        base_url = WatsonxOrchestrateTransformation.get_api_base_url(
+            cp4d_host, instance_id
+        )
+        auth_headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+        }
+        text = WatsonxOrchestrateTransformation.extract_text_from_a2a_params(params)
+        body = WatsonxOrchestrateTransformation.build_wxo_run_body(
+            wxo_agent_id=wxo_agent_id, text=text, thread_id=thread_id
+        )
+
+        try:
+            response = await client.post(
+                f"{base_url}/v1/orchestrate/runs/stream",
+                json=body,
+                headers=auth_headers,
+                stream=True,
+            )
+            response.raise_for_status()
+            content_type = response.headers.get("content-type", "").lower()
+
+            if "text/event-stream" not in content_type:
+                response_body = await response.aread()
+                result = json.loads(response_body)
+                accumulated_text = (
+                    WatsonxOrchestrateTransformation.extract_text_from_wxo_result(
+                        result
+                    )
+                )
+            else:
+                accumulated_text = (
+                    await WatsonxOrchestrateHandler._accumulate_wxo_sse_text(response)
+                )
+
+            async for (
+                chunk
+            ) in WatsonxOrchestrateTransformation.fake_streaming_from_text(
+                text=accumulated_text,
+                request_id=request_id,
+                chunk_size=chunk_size,
+                delay_ms=delay_ms,
+            ):
+                yield chunk
+
+        except httpx.TransportError as exc:
+            verbose_logger.warning(
+                f"WXO: Streaming transport failed ({exc!r}), "
+                "falling back to non-streaming + fake streaming",
+                exc_info=True,
+            )
+            result = await WatsonxOrchestrateHandler.handle_non_streaming(
+                request_id=request_id,
+                params=params,
+                litellm_params=litellm_params,
+            )
+            response_text = (
+                WatsonxOrchestrateTransformation.extract_text_from_a2a_message_response(
+                    result
+                )
+            )
+            async for (
+                chunk
+            ) in WatsonxOrchestrateTransformation.fake_streaming_from_text(
+                text=response_text,
+                request_id=request_id,
+                chunk_size=chunk_size,
+                delay_ms=delay_ms,
+            ):
+                yield chunk

diff --git a/litellm/a2a_protocol/providers/watsonx_orchestrate/transformation.py b/litellm/a2a_protocol/providers/watsonx_orchestrate/transformation.py
new file mode 100644
--- /dev/null
+++ b/litellm/a2a_protocol/providers/watsonx_orchestrate/transformation.py
@@ -1,0 +1,224 @@
+"""
+Transformation layer for IBM watsonx Orchestrate (WXO) agent provider.
+
+WXO uses a REST API (not A2A/JSON-RPC) with an async-poll execution model:
+  POST /v1/orchestrate/runs       → submit run, get run_id
+  GET  /v1/orchestrate/runs/{id}  → poll until terminal state
+  POST /v1/orchestrate/runs/stream → native SSE streaming
+"""
+
+import asyncio
+from typing import Any, AsyncIterator, Dict, Optional
+from uuid import uuid4
+
+from litellm._logging import verbose_logger
+
+
+class WatsonxOrchestrateTransformation:
+    """
+    Handles request/response transformation between A2A and the WXO REST API.
+    """
+
+    TERMINAL_STATES = frozenset(
+        {"completed", "succeeded", "failed", "error", "cancelled"}
+    )
+    SUCCESS_STATES = frozenset({"completed", "succeeded"})
+
+    @staticmethod
+    def get_api_base_url(cp4d_host: str, instance_id: str) -> str:
+        """Build the WXO API base URL from host and instance ID."""
+        return f"{cp4d_host.rstrip('/')}/orchestrate/cpd/instances/{instance_id}"
+
+    @staticmethod
+    def extract_text_from_a2a_params(params: Dict[str, Any]) -> str:
+        """
+        Extract user message text from A2A MessageSendParams.
+
+        A2A format: params.message.parts[*] where part.kind == "text"
+        """
+        message = params.get("message", {})
+        parts = message.get("parts", [])
+        texts = []
+        for part in parts:
+            if not isinstance(part, dict):
+                continue
+            kind = part.get("kind")
+            if kind in (None, "", "text") and part.get("text"):
+                texts.append(part["text"])
+        return " ".join(texts) or ""
+
+    @staticmethod
+    def build_wxo_run_body(
+        wxo_agent_id: str,
+        text: str,
+        thread_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Build the WXO POST /v1/orchestrate/runs request body."""
+        body: Dict[str, Any] = {
+            "agent_id": wxo_agent_id,
+            "message": {
+                "role": "user",
+                "content": [
+                    {
+                        "response_type": "conversational_search",
+                        "text": text,
+                    }
+                ],
+            },
+        }
+        if thread_id:
+            body["thread_id"] = thread_id
+        return body
+
+    @staticmethod
+    def extract_text_from_wxo_result(result: Any) -> str:
+        """
+        Extract response text from a WXO run result.
+
+        WXO can return text in several locations; checks in priority order per the API spec.
+        """
+        if not isinstance(result, dict):
+            return ""
+
+        # Primary: last_message.content[0].text
+        try:
+            text = result["last_message"]["content"][0]["text"]
+            if text:
+                return str(text)
+        except (KeyError, IndexError, TypeError):
+            pass
+
+        # Secondary: result.data.message.content[0].text
+        try:
+            text = result["result"]["data"]["message"]["content"][0]["text"]
+            if text:
+                return str(text)
+        except (KeyError, IndexError, TypeError):
+            pass
+
+        # Tertiary: results as a raw string
+        results = result.get("results")
+        if results and isinstance(results, str):
+            return results
+
+        return ""
+
+    @staticmethod
+    def extract_text_from_a2a_message_response(a2a_response: Dict[str, Any]) -> str:
+        result = a2a_response.get("result")
+        if not isinstance(result, dict):
+            verbose_logger.warning("WXO: A2A response missing result object")
+            return ""
+        parts = result.get("parts")
+        if not isinstance(parts, list):
+            verbose_logger.warning("WXO: A2A result has no parts list")
+            return ""
+        for part in parts:
+            if (
+                isinstance(part, dict)
+                and part.get("kind") == "text"
+                and part.get("text")
+            ):
+                return str(part["text"])
+        verbose_logger.warning("WXO: A2A result parts contained no text")
+        return ""
+
+    @staticmethod
+    def build_a2a_message_response(request_id: str, text: str) -> Dict[str, Any]:
+        """
+        Build a standard A2A non-streaming SendMessageResponse (kind=message).
+        """
+        return {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": {
+                "kind": "message",
+                "role": "agent",
+                "parts": [{"kind": "text", "text": text}],
+                "messageId": str(uuid4()),
+            },
+        }
+
+    @staticmethod
+    async def fake_streaming_from_text(
+        text: str,
+        request_id: str,
+        chunk_size: int = 50,
+        delay_ms: int = 10,
+    ) -> AsyncIterator[Dict[str, Any]]:
+        """
+        Emit standard A2A streaming events from a completed text response.
+
+        Event sequence:
+          1. task         (kind="task", state="submitted")
+          2. status-update (kind="status-update", state="working")
+          3. artifact-update chunks
+          4. status-update (kind="status-update", state="completed", final=True)
+        """
+        task_id = str(uuid4())
+        context_id = str(uuid4())
+        artifact_id = str(uuid4())
+
+        # 1. Task submitted
+        yield {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": {
+                "contextId": context_id,
+                "id": task_id,
+                "kind": "task",
+                "status": {"state": "submitted"},
+            },
+        }
+
+        # 2. Working
+        yield {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": {
+                "contextId": context_id,
+                "final": False,
+                "kind": "status-update",
+                "status": {"state": "working"},
+                "taskId": task_id,
+            },
+        }
+        await asyncio.sleep(delay_ms / 1000.0)
+
+        # 3. Artifact chunks (always emit at least one chunk, even for empty text)
+        text_to_chunk = text or ""
+        for i in range(0, max(len(text_to_chunk), 1), chunk_size):
+            chunk_text = text_to_chunk[i : i + chunk_size]
+            is_last = (i + chunk_size) >= max(len(text_to_chunk), 1)
+            yield {
+                "jsonrpc": "2.0",
+                "id": request_id,
+                "result": {
+                    "contextId": context_id,
+                    "kind": "artifact-update",
+                    "taskId": task_id,
+                    "artifact": {
+                        "artifactId": artifact_id,
+                        "parts": [{"kind": "text", "text": chunk_text}],
+                    },
+                },
+            }
+            if not is_last:
+                await asyncio.sleep(delay_ms / 1000.0)
+
+        # 4. Completed
+        yield {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": {
+                "contextId": context_id,
+                "final": True,
+                "kind": "status-update",
+                "status": {"state": "completed"},
+                "taskId": task_id,
+            },
+        }
+
+        verbose_logger.debug(
+            f"WXO: Fake streaming completed for request_id={request_id}"
+        )

diff --git a/litellm/proxy/public_endpoints/agent_create_fields.json b/litellm/proxy/public_endpoints/agent_create_fields.json
--- a/litellm/proxy/public_endpoints/agent_create_fields.json
+++ b/litellm/proxy/public_endpoints/agent_create_fields.json
@@ -189,6 +189,78 @@
     "litellm_params_template": {
       "custom_llm_provider": "vertex_ai"
     }
+  },
+  {
+    "agent_type": "watsonx_orchestrate",
+    "agent_type_display_name": "watsonx Orchestrate",
+    "description": "Connect to IBM watsonx Orchestrate agents via CP4D or IBM Cloud IAM",
+    "logo_url": "/ui/assets/logos/watsonx.svg",
+    "credential_fields": [
+      {
+        "key": "cp4d_host",
+        "label": "CP4D Host URL",
+        "placeholder": "https://cpd-cpd.apps.example.com",
+        "tooltip": "Your CP4D cluster base URL (e.g. https://cpd-cpd.apps.example.com). For IBM Cloud WXO, use the service endpoint.",
+        "required": true,
+        "field_type": "text",
+        "default_value": null,
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "instance_id",
+        "label": "WXO Instance ID",
+        "placeholder": "1769134113217795",
+        "tooltip": "The numeric watsonx Orchestrate instance ID. Find it in the WXO service URL: /orchestrate/cpd/instances/<INSTANCE_ID>",
+        "required": true,
+        "field_type": "text",
+        "default_value": null,
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "wxo_agent_id",
+        "label": "WXO Agent ID",
+        "placeholder": "588c8cdf-60f4-454b-8468-8702b19dca46",
+        "tooltip": "UUID of the agent in watsonx Orchestrate. Find it via the WXO console or GET /v1/orchestrate/agents.",
+        "required": true,
+        "field_type": "text",
+        "default_value": null,
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "auth_mode",
+        "label": "Authentication Mode",
+        "placeholder": null,
+        "tooltip": "cp4d: on-prem / CloudPak for Data (requires username). ibm_cloud: IBM Cloud IAM (api_key only).",
+        "required": false,
+        "field_type": "select",
+        "options": ["cp4d", "ibm_cloud"],
+        "default_value": "cp4d",
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "username",
+        "label": "Username (CP4D only)",
+        "placeholder": "admin",
+        "tooltip": "Your CP4D username. Required when auth_mode is 'cp4d'.",
+        "required": false,
+        "field_type": "text",
+        "default_value": null,
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "api_key",
+        "label": "API Key",
+        "placeholder": null,
+        "tooltip": "CP4D API key (auth_mode=cp4d) or IBM Cloud API key (auth_mode=ibm_cloud).",
+        "required": true,
+        "field_type": "password",
+        "default_value": null,
+        "include_in_litellm_params": true
+      }
+    ],
+    "litellm_params_template": {
+      "custom_llm_provider": "watsonx_orchestrate"
+    }
   }
 ]
 

diff --git a/tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py b/tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py
new file mode 100644
--- /dev/null
+++ b/tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py
@@ -1,0 +1,265 @@
+import json
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../../../../.."))
+
+from litellm.a2a_protocol.providers.config_manager import A2AProviderConfigManager
+from litellm.a2a_protocol.providers.watsonx_orchestrate.handler import (
+    WatsonxOrchestrateHandler,
+)
+from litellm.a2a_protocol.providers.watsonx_orchestrate.transformation import (
+    WatsonxOrchestrateTransformation,
+)
+
+
+class _JsonResponse:
+    def __init__(self, payload):
+        self.payload = payload
+
+    def raise_for_status(self):
+        pass
+
+    def json(self):
+        return self.payload
+
+
+class _ShortTtlTokenClient:
+    def __init__(self):
... diff truncated: showing 800 of 1035 lines

_{You can send follow-ups to the cloud agent here.}

[Sameerlite]

bugbot run

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 2 potential issues.

Autofix Details

Bugbot Autofix prepared fixes for both issues found in the latest run.

• ✅ Fixed: Streaming non-SSE path skips status check and polling
  • Added shared WXO run completion validation so non-SSE streaming JSON polls pending runs and raises on failed terminal states before extracting text.
• ✅ Fixed: Default auth mode requires username marked optional in dashboard
  • Marked the dashboard username field required to match the default cp4d authentication mode runtime requirement.

Preview (4306dd1630)

diff --git a/litellm/a2a_protocol/providers/config_manager.py b/litellm/a2a_protocol/providers/config_manager.py
--- a/litellm/a2a_protocol/providers/config_manager.py
+++ b/litellm/a2a_protocol/providers/config_manager.py
@@ -48,4 +48,11 @@
 
             return BedrockAgentCoreA2AConfig()
 
+        if custom_llm_provider == "watsonx_orchestrate":
+            from litellm.a2a_protocol.providers.watsonx_orchestrate.config import (
+                WatsonxOrchestrateA2AConfig,
+            )
+
+            return WatsonxOrchestrateA2AConfig()
+
         return None

diff --git a/litellm/a2a_protocol/providers/watsonx_orchestrate/__init__.py b/litellm/a2a_protocol/providers/watsonx_orchestrate/__init__.py
new file mode 100644
--- /dev/null
+++ b/litellm/a2a_protocol/providers/watsonx_orchestrate/__init__.py
@@ -1,0 +1,3 @@
+"""
+IBM watsonx Orchestrate (WXO) A2A provider.
+"""

diff --git a/litellm/a2a_protocol/providers/watsonx_orchestrate/config.py b/litellm/a2a_protocol/providers/watsonx_orchestrate/config.py
new file mode 100644
--- /dev/null
+++ b/litellm/a2a_protocol/providers/watsonx_orchestrate/config.py
@@ -1,0 +1,55 @@
+"""
+A2A provider configuration for IBM watsonx Orchestrate (WXO).
+"""
+
+from typing import Any, AsyncIterator, Dict, Optional
+
+from litellm.a2a_protocol.providers.base import BaseA2AProviderConfig
+from litellm.a2a_protocol.providers.watsonx_orchestrate.handler import (
+    WatsonxOrchestrateHandler,
+)
+
+
+class WatsonxOrchestrateA2AConfig(BaseA2AProviderConfig):
+    """A2A bridge for IBM watsonx Orchestrate (REST runs API + poll/SSE)."""
+
+    async def handle_non_streaming(
+        self,
+        request_id: str,
+        params: Dict[str, Any],
+        api_base: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """Handle a non-streaming A2A request via WXO runs API."""
+        litellm_params = kwargs.get("litellm_params")
+        if not litellm_params:
+            raise ValueError(
+                "litellm_params is required for WatsonxOrchestrateA2AConfig "
+                "(must contain cp4d_host, instance_id, wxo_agent_id, api_key)"
+            )
+        return await WatsonxOrchestrateHandler.handle_non_streaming(
+            request_id=request_id,
+            params=params,
+            litellm_params=litellm_params,
+        )
+
+    async def handle_streaming(
+        self,
+        request_id: str,
+        params: Dict[str, Any],
+        api_base: Optional[str] = None,
+        **kwargs: Any,
+    ) -> AsyncIterator[Dict[str, Any]]:
+        """Handle a streaming A2A request via WXO streaming runs API."""
+        litellm_params = kwargs.get("litellm_params")
+        if not litellm_params:
+            raise ValueError(
+                "litellm_params is required for WatsonxOrchestrateA2AConfig "
+                "(must contain cp4d_host, instance_id, wxo_agent_id, api_key)"
+            )
+        async for chunk in WatsonxOrchestrateHandler.handle_streaming(
+            request_id=request_id,
+            params=params,
+            litellm_params=litellm_params,
+        ):
+            yield chunk

diff --git a/litellm/a2a_protocol/providers/watsonx_orchestrate/handler.py b/litellm/a2a_protocol/providers/watsonx_orchestrate/handler.py
new file mode 100644
--- /dev/null
+++ b/litellm/a2a_protocol/providers/watsonx_orchestrate/handler.py
@@ -1,0 +1,382 @@
+"""
+Handler for IBM watsonx Orchestrate (WXO) agent provider.
+"""
+
+import asyncio
+import hashlib
+import json
+import time
+from typing import Any, AsyncIterator, Dict, Optional, Tuple, cast
+
+import httpx
+
+from litellm._logging import verbose_logger
+from litellm.a2a_protocol.providers.watsonx_orchestrate.transformation import (
+    WatsonxOrchestrateTransformation,
+)
+from litellm.llms.custom_httpx.http_handler import (
+    AsyncHTTPHandler,
+    get_async_httpx_client,
+)
+from litellm.types.llms.custom_http import httpxSpecialProvider
+
+_IBM_CLOUD_IAM_URL = "https://iam.cloud.ibm.com/identity/token"
+_POLL_INTERVAL_S = 2.0
+_MAX_POLL_ATTEMPTS = 90
+_TOKEN_CACHE_TTL_BUFFER_S = 60
+_token_cache: Dict[str, Tuple[str, float]] = {}
+
+
+class WatsonxOrchestrateHandler:
+    @staticmethod
+    def _http_client(timeout: float = 90.0) -> AsyncHTTPHandler:
+        return get_async_httpx_client(
+            llm_provider=cast(Any, httpxSpecialProvider.A2AProvider),
+            params={"timeout": timeout},
+        )
+
+    @staticmethod
+    def _token_cache_key(
+        auth_mode: str,
+        cp4d_host: str,
+        api_key: str,
+        username: Optional[str],
+    ) -> str:
+        material = f"{auth_mode}:{cp4d_host}:{username or ''}:{api_key}"
+        return hashlib.sha256(material.encode()).hexdigest()
+
+    @staticmethod
+    def _cp4d_token_ttl_seconds(
+        expiration: Any, now_wall: Optional[float] = None
+    ) -> int:
+        # CP4D returns expiration as absolute Unix epoch seconds, not a duration.
+        expires_at = int(expiration)
+        wall = now_wall if now_wall is not None else time.time()
+        return max(expires_at - int(wall), 0)
+
+    @staticmethod
+    async def _get_bearer_token(
+        cp4d_host: str,
+        auth_mode: str,
+        api_key: str,
+        username: Optional[str] = None,
+        client: Optional[AsyncHTTPHandler] = None,
+    ) -> str:
+        cache_key = WatsonxOrchestrateHandler._token_cache_key(
+            auth_mode, cp4d_host, api_key, username
+        )
+        now = time.monotonic()
+        cached = _token_cache.get(cache_key)
+        if cached and cached[1] > now:
+            return cached[0]
+
+        if client is None:
+            client = WatsonxOrchestrateHandler._http_client(timeout=30.0)
+
+        if auth_mode == "ibm_cloud":
+            response = await client.post(
+                _IBM_CLOUD_IAM_URL,
+                data={
+                    "grant_type": "urn:ibm:params:oauth:grant-type:apikey",
+                    "apikey": api_key,
+                },
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+            )
+            response.raise_for_status()
+            payload = response.json()
+            token = str(payload["access_token"])
+            ttl_s = int(payload.get("expires_in", 3600))
+        else:
+            if not username:
+                raise ValueError(
+                    "'username' is required in litellm_params when auth_mode='cp4d'"
+                )
+            token_url = f"{cp4d_host.rstrip('/')}/icp4d-api/v1/authorize"
+            response = await client.post(
+                token_url,
+                json={"username": username, "api_key": api_key},
+                headers={"Content-Type": "application/json"},
+            )
+            response.raise_for_status()
+            payload = response.json()
+            token = str(payload["token"])
+            expiration = payload.get("expiration")
+            if expiration is None:
+                ttl_s = 3600
+            else:
+                ttl_s = WatsonxOrchestrateHandler._cp4d_token_ttl_seconds(expiration)
+
+        expires_at = now + max(ttl_s - _TOKEN_CACHE_TTL_BUFFER_S, 0)
+        _token_cache[cache_key] = (token, expires_at)
+        return token
+
+    @staticmethod
+    async def _poll_run(
+        base_url: str,
+        run_id: str,
+        auth_headers: Dict[str, str],
+        client: AsyncHTTPHandler,
+        max_attempts: int = _MAX_POLL_ATTEMPTS,
+        interval_s: float = _POLL_INTERVAL_S,
+    ) -> Dict[str, Any]:
+        url = f"{base_url}/v1/orchestrate/runs/{run_id}"
+
+        for attempt in range(max_attempts):
+            await asyncio.sleep(interval_s)
+            response = await client.get(url, headers=auth_headers)
+            response.raise_for_status()
+            result: Dict[str, Any] = response.json()
+            status = result.get("status", "")
+            verbose_logger.debug(
+                f"WXO: Poll {attempt + 1}/{max_attempts} run='{run_id}' status='{status}'"
+            )
+            if status in WatsonxOrchestrateTransformation.TERMINAL_STATES:
+                return result
+
+        raise TimeoutError(
+            f"WXO run '{run_id}' did not reach a terminal state after "
+            f"{max_attempts * interval_s:.0f}s"
+        )
+
+    @staticmethod
+    async def _get_successful_run_data(
+        run_data: Dict[str, Any],
+        base_url: str,
+        auth_headers: Dict[str, str],
+        client: AsyncHTTPHandler,
+    ) -> Dict[str, Any]:
+        status = run_data.get("status", "")
+        if status not in WatsonxOrchestrateTransformation.TERMINAL_STATES:
+            run_id = run_data.get("run_id") or run_data.get("id") or ""
+            if not run_id:
+                raise ValueError(f"WXO: No run_id in response: {run_data}")
+            run_data = await WatsonxOrchestrateHandler._poll_run(
+                base_url=base_url,
+                run_id=run_id,
+                auth_headers=auth_headers,
+                client=client,
+            )
+            status = run_data.get("status", "")
+
+        if status not in WatsonxOrchestrateTransformation.SUCCESS_STATES:
+            raise RuntimeError(
+                f"WXO run ended with non-success status '{status}': {run_data}"
+            )
+
+        return run_data
+
+    @staticmethod
+    async def _accumulate_wxo_sse_text(response: Any) -> str:
+        accumulated_text = ""
+        async for line in response.aiter_lines():
+            if not line.startswith("data:"):
+                continue
+            data_str = line[5:].strip()
+            if not data_str or data_str == "[DONE]":
+                continue
+            try:
+                event = json.loads(data_str)
+            except json.JSONDecodeError:
+                continue
+            chunk_text = WatsonxOrchestrateTransformation.extract_text_from_wxo_result(
+                event
+            )
+            if chunk_text:
+                accumulated_text += chunk_text
+        return accumulated_text
+
+    @staticmethod
+    def _extract_litellm_params(litellm_params: Dict[str, Any]) -> tuple:
+        cp4d_host = litellm_params.get("cp4d_host") or ""
+        instance_id = litellm_params.get("instance_id") or ""
+        wxo_agent_id = litellm_params.get("wxo_agent_id") or ""
+        api_key = litellm_params.get("api_key") or ""
+        username = litellm_params.get("username") or None
+        auth_mode = litellm_params.get("auth_mode") or "cp4d"
+        thread_id = litellm_params.get("thread_id") or None
+
+        if not cp4d_host:
+            raise ValueError("'cp4d_host' is required in litellm_params for WXO agents")
+        if not instance_id:
+            raise ValueError(
+                "'instance_id' is required in litellm_params for WXO agents"
+            )
+        if not wxo_agent_id:
+            raise ValueError(
+                "'wxo_agent_id' is required in litellm_params for WXO agents"
+            )
+        if not api_key:
+            raise ValueError("'api_key' is required in litellm_params for WXO agents")
+
+        return (
+            cp4d_host,
+            instance_id,
+            wxo_agent_id,
+            api_key,
+            username,
+            auth_mode,
+            thread_id,
+        )
+
+    @staticmethod
+    async def handle_non_streaming(
+        request_id: str,
+        params: Dict[str, Any],
+        litellm_params: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        (
+            cp4d_host,
+            instance_id,
+            wxo_agent_id,
+            api_key,
+            username,
+            auth_mode,
+            thread_id,
+        ) = WatsonxOrchestrateHandler._extract_litellm_params(litellm_params)
+
+        client = WatsonxOrchestrateHandler._http_client(timeout=90.0)
+        token = await WatsonxOrchestrateHandler._get_bearer_token(
+            cp4d_host=cp4d_host,
+            auth_mode=auth_mode,
+            api_key=api_key,
+            username=username,
+            client=client,
+        )
+        base_url = WatsonxOrchestrateTransformation.get_api_base_url(
+            cp4d_host, instance_id
+        )
+        auth_headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        }
+
+        text = WatsonxOrchestrateTransformation.extract_text_from_a2a_params(params)
+        body = WatsonxOrchestrateTransformation.build_wxo_run_body(
+            wxo_agent_id=wxo_agent_id, text=text, thread_id=thread_id
+        )
+
+        run_response = await client.post(
+            f"{base_url}/v1/orchestrate/runs",
+            json=body,
+            headers=auth_headers,
+        )
+        run_response.raise_for_status()
+        run_data: Dict[str, Any] = run_response.json()
+
+        run_data = await WatsonxOrchestrateHandler._get_successful_run_data(
+            run_data=run_data,
+            base_url=base_url,
+            auth_headers=auth_headers,
+            client=client,
+        )
+
+        response_text = WatsonxOrchestrateTransformation.extract_text_from_wxo_result(
+            run_data
+        )
+        return WatsonxOrchestrateTransformation.build_a2a_message_response(
+            request_id=request_id, text=response_text
+        )
+
+    @staticmethod
+    async def handle_streaming(
+        request_id: str,
+        params: Dict[str, Any],
+        litellm_params: Dict[str, Any],
+        chunk_size: int = 50,
+        delay_ms: int = 10,
+    ) -> AsyncIterator[Dict[str, Any]]:
+        (
+            cp4d_host,
+            instance_id,
+            wxo_agent_id,
+            api_key,
+            username,
+            auth_mode,
+            thread_id,
+        ) = WatsonxOrchestrateHandler._extract_litellm_params(litellm_params)
+
+        client = WatsonxOrchestrateHandler._http_client(timeout=120.0)
+        token = await WatsonxOrchestrateHandler._get_bearer_token(
+            cp4d_host=cp4d_host,
+            auth_mode=auth_mode,
+            api_key=api_key,
+            username=username,
+            client=client,
+        )
+        base_url = WatsonxOrchestrateTransformation.get_api_base_url(
+            cp4d_host, instance_id
+        )
+        auth_headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+        }
+        text = WatsonxOrchestrateTransformation.extract_text_from_a2a_params(params)
+        body = WatsonxOrchestrateTransformation.build_wxo_run_body(
+            wxo_agent_id=wxo_agent_id, text=text, thread_id=thread_id
+        )
+
+        try:
+            response = await client.post(
+                f"{base_url}/v1/orchestrate/runs/stream",
+                json=body,
+                headers=auth_headers,
+                stream=True,
+            )
+            response.raise_for_status()
+            content_type = response.headers.get("content-type", "").lower()
+
+            if "text/event-stream" not in content_type:
+                response_body = await response.aread()
+                result = json.loads(response_body)
+                result = await WatsonxOrchestrateHandler._get_successful_run_data(
+                    run_data=result,
+                    base_url=base_url,
+                    auth_headers=auth_headers,
+                    client=client,
+                )
+                accumulated_text = (
+                    WatsonxOrchestrateTransformation.extract_text_from_wxo_result(
+                        result
+                    )
+                )
+            else:
+                accumulated_text = (
+                    await WatsonxOrchestrateHandler._accumulate_wxo_sse_text(response)
+                )
+
+            async for (
+                chunk
+            ) in WatsonxOrchestrateTransformation.fake_streaming_from_text(
+                text=accumulated_text,
+                request_id=request_id,
+                chunk_size=chunk_size,
+                delay_ms=delay_ms,
+            ):
+                yield chunk
+
+        except httpx.TransportError as exc:
+            verbose_logger.warning(
+                f"WXO: Streaming transport failed ({exc!r}), "
+                "falling back to non-streaming + fake streaming",
+                exc_info=True,
+            )
+            result = await WatsonxOrchestrateHandler.handle_non_streaming(
+                request_id=request_id,
+                params=params,
+                litellm_params=litellm_params,
+            )
+            response_text = (
+                WatsonxOrchestrateTransformation.extract_text_from_a2a_message_response(
+                    result
+                )
+            )
+            async for (
+                chunk
+            ) in WatsonxOrchestrateTransformation.fake_streaming_from_text(
+                text=response_text,
+                request_id=request_id,
+                chunk_size=chunk_size,
+                delay_ms=delay_ms,
+            ):
+                yield chunk

diff --git a/litellm/a2a_protocol/providers/watsonx_orchestrate/transformation.py b/litellm/a2a_protocol/providers/watsonx_orchestrate/transformation.py
new file mode 100644
--- /dev/null
+++ b/litellm/a2a_protocol/providers/watsonx_orchestrate/transformation.py
@@ -1,0 +1,224 @@
+"""
+Transformation layer for IBM watsonx Orchestrate (WXO) agent provider.
+
+WXO uses a REST API (not A2A/JSON-RPC) with an async-poll execution model:
+  POST /v1/orchestrate/runs       → submit run, get run_id
+  GET  /v1/orchestrate/runs/{id}  → poll until terminal state
+  POST /v1/orchestrate/runs/stream → native SSE streaming
+"""
+
+import asyncio
+from typing import Any, AsyncIterator, Dict, Optional
+from uuid import uuid4
+
+from litellm._logging import verbose_logger
+
+
+class WatsonxOrchestrateTransformation:
+    """
+    Handles request/response transformation between A2A and the WXO REST API.
+    """
+
+    TERMINAL_STATES = frozenset(
+        {"completed", "succeeded", "failed", "error", "cancelled"}
+    )
+    SUCCESS_STATES = frozenset({"completed", "succeeded"})
+
+    @staticmethod
+    def get_api_base_url(cp4d_host: str, instance_id: str) -> str:
+        """Build the WXO API base URL from host and instance ID."""
+        return f"{cp4d_host.rstrip('/')}/orchestrate/cpd/instances/{instance_id}"
+
+    @staticmethod
+    def extract_text_from_a2a_params(params: Dict[str, Any]) -> str:
+        """
+        Extract user message text from A2A MessageSendParams.
+
+        A2A format: params.message.parts[*] where part.kind == "text"
+        """
+        message = params.get("message", {})
+        parts = message.get("parts", [])
+        texts = []
+        for part in parts:
+            if not isinstance(part, dict):
+                continue
+            kind = part.get("kind")
+            if kind in (None, "", "text") and part.get("text"):
+                texts.append(part["text"])
+        return " ".join(texts) or ""
+
+    @staticmethod
+    def build_wxo_run_body(
+        wxo_agent_id: str,
+        text: str,
+        thread_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Build the WXO POST /v1/orchestrate/runs request body."""
+        body: Dict[str, Any] = {
+            "agent_id": wxo_agent_id,
+            "message": {
+                "role": "user",
+                "content": [
+                    {
+                        "response_type": "conversational_search",
+                        "text": text,
+                    }
+                ],
+            },
+        }
+        if thread_id:
+            body["thread_id"] = thread_id
+        return body
+
+    @staticmethod
+    def extract_text_from_wxo_result(result: Any) -> str:
+        """
+        Extract response text from a WXO run result.
+
+        WXO can return text in several locations; checks in priority order per the API spec.
+        """
+        if not isinstance(result, dict):
+            return ""
+
+        # Primary: last_message.content[0].text
+        try:
+            text = result["last_message"]["content"][0]["text"]
+            if text:
+                return str(text)
+        except (KeyError, IndexError, TypeError):
+            pass
+
+        # Secondary: result.data.message.content[0].text
+        try:
+            text = result["result"]["data"]["message"]["content"][0]["text"]
+            if text:
+                return str(text)
+        except (KeyError, IndexError, TypeError):
+            pass
+
+        # Tertiary: results as a raw string
+        results = result.get("results")
+        if results and isinstance(results, str):
+            return results
+
+        return ""
+
+    @staticmethod
+    def extract_text_from_a2a_message_response(a2a_response: Dict[str, Any]) -> str:
+        result = a2a_response.get("result")
+        if not isinstance(result, dict):
+            verbose_logger.warning("WXO: A2A response missing result object")
+            return ""
+        parts = result.get("parts")
+        if not isinstance(parts, list):
+            verbose_logger.warning("WXO: A2A result has no parts list")
+            return ""
+        for part in parts:
+            if (
+                isinstance(part, dict)
+                and part.get("kind") == "text"
+                and part.get("text")
+            ):
+                return str(part["text"])
+        verbose_logger.warning("WXO: A2A result parts contained no text")
+        return ""
+
+    @staticmethod
+    def build_a2a_message_response(request_id: str, text: str) -> Dict[str, Any]:
+        """
+        Build a standard A2A non-streaming SendMessageResponse (kind=message).
+        """
+        return {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": {
+                "kind": "message",
+                "role": "agent",
+                "parts": [{"kind": "text", "text": text}],
+                "messageId": str(uuid4()),
+            },
+        }
+
+    @staticmethod
+    async def fake_streaming_from_text(
+        text: str,
+        request_id: str,
+        chunk_size: int = 50,
+        delay_ms: int = 10,
+    ) -> AsyncIterator[Dict[str, Any]]:
+        """
+        Emit standard A2A streaming events from a completed text response.
+
+        Event sequence:
+          1. task         (kind="task", state="submitted")
+          2. status-update (kind="status-update", state="working")
+          3. artifact-update chunks
+          4. status-update (kind="status-update", state="completed", final=True)
+        """
+        task_id = str(uuid4())
+        context_id = str(uuid4())
+        artifact_id = str(uuid4())
+
+        # 1. Task submitted
+        yield {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": {
+                "contextId": context_id,
+                "id": task_id,
+                "kind": "task",
+                "status": {"state": "submitted"},
+            },
+        }
+
+        # 2. Working
+        yield {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": {
+                "contextId": context_id,
+                "final": False,
+                "kind": "status-update",
+                "status": {"state": "working"},
+                "taskId": task_id,
+            },
+        }
+        await asyncio.sleep(delay_ms / 1000.0)
+
+        # 3. Artifact chunks (always emit at least one chunk, even for empty text)
+        text_to_chunk = text or ""
+        for i in range(0, max(len(text_to_chunk), 1), chunk_size):
+            chunk_text = text_to_chunk[i : i + chunk_size]
+            is_last = (i + chunk_size) >= max(len(text_to_chunk), 1)
+            yield {
+                "jsonrpc": "2.0",
+                "id": request_id,
+                "result": {
+                    "contextId": context_id,
+                    "kind": "artifact-update",
+                    "taskId": task_id,
+                    "artifact": {
+                        "artifactId": artifact_id,
+                        "parts": [{"kind": "text", "text": chunk_text}],
+                    },
+                },
+            }
+            if not is_last:
+                await asyncio.sleep(delay_ms / 1000.0)
+
+        # 4. Completed
+        yield {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": {
+                "contextId": context_id,
+                "final": True,
+                "kind": "status-update",
+                "status": {"state": "completed"},
+                "taskId": task_id,
+            },
+        }
+
+        verbose_logger.debug(
+            f"WXO: Fake streaming completed for request_id={request_id}"
+        )

diff --git a/litellm/proxy/public_endpoints/agent_create_fields.json b/litellm/proxy/public_endpoints/agent_create_fields.json
--- a/litellm/proxy/public_endpoints/agent_create_fields.json
+++ b/litellm/proxy/public_endpoints/agent_create_fields.json
@@ -189,6 +189,78 @@
     "litellm_params_template": {
       "custom_llm_provider": "vertex_ai"
     }
+  },
+  {
+    "agent_type": "watsonx_orchestrate",
+    "agent_type_display_name": "watsonx Orchestrate",
+    "description": "Connect to IBM watsonx Orchestrate agents via CP4D or IBM Cloud IAM",
+    "logo_url": "/ui/assets/logos/watsonx.svg",
+    "credential_fields": [
+      {
+        "key": "cp4d_host",
+        "label": "CP4D Host URL",
+        "placeholder": "https://cpd-cpd.apps.example.com",
+        "tooltip": "Your CP4D cluster base URL (e.g. https://cpd-cpd.apps.example.com). For IBM Cloud WXO, use the service endpoint.",
+        "required": true,
+        "field_type": "text",
+        "default_value": null,
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "instance_id",
+        "label": "WXO Instance ID",
+        "placeholder": "1769134113217795",
+        "tooltip": "The numeric watsonx Orchestrate instance ID. Find it in the WXO service URL: /orchestrate/cpd/instances/<INSTANCE_ID>",
+        "required": true,
+        "field_type": "text",
+        "default_value": null,
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "wxo_agent_id",
+        "label": "WXO Agent ID",
+        "placeholder": "588c8cdf-60f4-454b-8468-8702b19dca46",
+        "tooltip": "UUID of the agent in watsonx Orchestrate. Find it via the WXO console or GET /v1/orchestrate/agents.",
+        "required": true,
+        "field_type": "text",
+        "default_value": null,
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "auth_mode",
+        "label": "Authentication Mode",
+        "placeholder": null,
+        "tooltip": "cp4d: on-prem / CloudPak for Data (requires username). ibm_cloud: IBM Cloud IAM (api_key only).",
+        "required": false,
+        "field_type": "select",
+        "options": ["cp4d", "ibm_cloud"],
+        "default_value": "cp4d",
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "username",
+        "label": "Username (CP4D only)",
+        "placeholder": "admin",
+        "tooltip": "Your CP4D username. Required when auth_mode is 'cp4d'.",
+        "required": true,
+        "field_type": "text",
+        "default_value": null,
+        "include_in_litellm_params": true
+      },
+      {
+        "key": "api_key",
+        "label": "API Key",
+        "placeholder": null,
+        "tooltip": "CP4D API key (auth_mode=cp4d) or IBM Cloud API key (auth_mode=ibm_cloud).",
+        "required": true,
+        "field_type": "password",
+        "default_value": null,
+        "include_in_litellm_params": true
+      }
+    ],
+    "litellm_params_template": {
+      "custom_llm_provider": "watsonx_orchestrate"
+    }
   }
 ]
 

diff --git a/tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py b/tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py
new file mode 100644
--- /dev/null
+++ b/tests/test_litellm/a2a_protocol/providers/watsonx_orchestrate/test_watsonx_orchestrate_transformation.py
@@ -1,0 +1,393 @@
+import json
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../../../../.."))
... diff truncated: showing 800 of 1185 lines

_{You can send follow-ups to the cloud agent here.}

[Sameerlite]

bugbot run

[Sameerlite]

@greptileai

[Sameerlite]

bugbot run

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

^{Reviewed by Cursor Bugbot for commit e034cec. Configure here.}

[Sameerlite]

@greptileai

[Sameerlite]

bugbot run

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

^{Reviewed by Cursor Bugbot for commit 900926a. Configure here.}

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[mateo-berri]

LGTM; thanks!