feat: add caller-controlled MCP methods

src/otari/__init__.py

@@ -30,6 +30,8 @@
     BatchNotCompleteError,
     GatewayTimeoutError,
     InsufficientFundsError,
+    MCPError,
+    MCPOutcomeUnknownError,
     ModelNotFoundError,
     OtariError,
     RateLimitError,
@@ -42,11 +44,18 @@
     BatchResult,
     BatchResultError,
     BatchResultItem,
+    CallToolResult,
     ChatCompletion,
     ChatCompletionChunk,
     CreateBatchParams,
     CreateEmbeddingResponse,
+    ExecutionState,
     ListBatchesOptions,
+    McpErrorBody,
+    McpExecuteRequest,
+    McpToolDefinition,
+    McpToolsResponse,
+    McpToolWarning,
     MessageResponse,
     ModelObject,
     ModerationResponse,
@@ -70,14 +79,23 @@
     "BatchResult",
     "BatchResultError",
     "BatchResultItem",
+    "CallToolResult",
     "ChatCompletion",
     "ChatCompletionChunk",
     "ControlPlane",
     "CreateBatchParams",
     "CreateEmbeddingResponse",
+    "ExecutionState",
     "GatewayTimeoutError",
     "InsufficientFundsError",
     "ListBatchesOptions",
+    "MCPError",
+    "MCPOutcomeUnknownError",
+    "McpErrorBody",
+    "McpExecuteRequest",
+    "McpToolDefinition",
+    "McpToolWarning",
+    "McpToolsResponse",
     "MessageResponse",
     "ModelNotFoundError",
     "ModelObject",

src/otari/async_client.py

@@ -54,6 +54,7 @@
 from otari._streaming import aiter_sse
 from otari.control_plane import ControlPlane
 from otari.errors import OtariError
+from otari.mcp import AsyncMCP
 from otari.response_metadata import AsyncOtariStream, OtariResponse
 
 if TYPE_CHECKING:
@@ -97,7 +98,7 @@ class AsyncOtariClient(_BaseOtariClient):
             ``GATEWAY_PLATFORM_TOKEN``).
         admin_key: Master/admin key for the control-plane (``GATEWAY_ADMIN_KEY``).
         default_headers: Additional default headers sent with every request.
-        timeout: Per-request timeout (seconds) for the streaming shim.
+        timeout: Per-request timeout (seconds) for streaming and MCP requests.
     """
 
     def __init__(
@@ -124,6 +125,8 @@ def __init__(
         api_any = cast("Any", self._api)
         for name, value in self._default_headers.items():
             api_any.set_default_header(name, value)
+        # HTTPX defaults to retries=0, including environment proxy transports.
+        # MCP's lower-level attempt-count tests guard this safety property.
         self._http = httpx.AsyncClient(timeout=timeout)
 
         self._chat = ChatApi(self._api)
@@ -136,6 +139,11 @@ def __init__(
         self._images = ImagesApi(self._api)
         self._batches = BatchesApi(self._api)
 
+    @cached_property
+    def mcp(self) -> AsyncMCP:
+        """Stored-server tool discovery and single-attempt authorized execution."""
+        return AsyncMCP(self)
+
     @cached_property
     def control_plane(self) -> ControlPlane:
         """Typed client for the management endpoints (keys, users, budgets, pricing, usage).

src/otari/client.py

@@ -54,6 +54,7 @@
 from otari._streaming import iter_sse
 from otari.control_plane import ControlPlane
 from otari.errors import OtariError
+from otari.mcp import MCP
 from otari.response_metadata import OtariResponse, OtariStream
 
 if TYPE_CHECKING:
@@ -101,7 +102,7 @@ class OtariClient(_BaseOtariClient):
         admin_key: Master/admin key for the control-plane endpoints. Falls back
             to ``GATEWAY_ADMIN_KEY`` (or the platform token in platform mode).
         default_headers: Additional default headers sent with every request.
-        timeout: Per-request timeout (seconds) for the streaming shim.
+        timeout: Per-request timeout (seconds) for streaming and MCP requests.
     """
 
     def __init__(
@@ -128,8 +129,9 @@ def __init__(
         api_any = cast("Any", self._api)
         for name, value in self._default_headers.items():
             api_any.set_default_header(name, value)
-        # Raw httpx client used only for the SSE streaming shim (the generated
-        # core buffers and cannot stream).
+        # HTTPX defaults to retries=0 (including environment proxy transports).
+        # MCP tests count attempts below httpcore's retry loop. Keep the default
+        # transport: supplying a custom one disables environment proxy discovery.
         self._http = httpx.Client(timeout=timeout)
 
         self._chat = ChatApi(self._api)
@@ -142,6 +144,11 @@ def __init__(
         self._images = ImagesApi(self._api)
         self._batches = BatchesApi(self._api)
 
+    @cached_property
+    def mcp(self) -> MCP:
+        """Stored-server tool discovery and single-attempt authorized execution."""
+        return MCP(self)
+
     @cached_property
     def control_plane(self) -> ControlPlane:
         """Typed client for the management endpoints (keys, users, budgets, pricing, usage).

src/otari/errors.py

@@ -7,6 +7,8 @@
 
 from __future__ import annotations
 
+from typing import Literal
+
 
 class OtariError(Exception):
     """Base exception for all otari errors.
@@ -40,6 +42,36 @@ def __str__(self) -> str:
         return self.message
 
 
+class MCPError(OtariError):
+    """Caller-orchestrated MCP failure, distinct from inference/batch errors.
+
+    ``code`` is absent without a typed gateway response; ``request_id`` may
+    still be available from its response header.
+    ``execution_state`` is conservative: ``outcome_unknown`` means the remote
+    tool may already have run. Neither state triggers an SDK retry.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        execution_state: Literal["not_started", "outcome_unknown"],
+        code: str | None = None,
+        request_id: str | None = None,
+        status_code: int | None = None,
+        retry_after: str | None = None,
+    ) -> None:
+        super().__init__(message, status_code=status_code, provider_name="gateway")
+        self.code = code
+        self.execution_state = execution_state
+        self.request_id = request_id
+        self.retry_after = retry_after
+
+
+class MCPOutcomeUnknownError(MCPError):
+    """Execution may have occurred. Do not retry or fall back to direct MCP."""
+
+
 class AuthenticationError(OtariError):
     """Raised when authentication with the gateway fails (HTTP 401, 403)."""
 

src/otari/mcp.py

@@ -0,0 +1,189 @@
+"""Caller-controlled MCP, using generated models but a single-attempt transport.
+
+The generated urllib3 client inherits connection retries unless overridden.
+Use the client's HTTPX transport instead: zero connection retries, no status
+retries, no redirects, and no inference error mapping (a 409 here is not a
+batch error). The owning Otari client closes this shared transport.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+from uuid import UUID
+
+import httpx
+
+from otari._base import build_request
+from otari.errors import MCPError, MCPOutcomeUnknownError
+from otari.types import CallToolResult, McpErrorBody, McpExecuteRequest, McpToolsResponse
+
+if TYPE_CHECKING:
+    from otari.async_client import AsyncOtariClient
+    from otari.client import OtariClient
+
+
+def _unknown_response(response: httpx.Response | None = None) -> MCPOutcomeUnknownError:
+    # Never include raw response/exception text, which may contain call data.
+    return MCPOutcomeUnknownError(
+        "MCP execution outcome is unknown; the remote tool may already have run",
+        execution_state="outcome_unknown",
+        status_code=response.status_code if response is not None else None,
+        request_id=response.headers.get("X-Otari-Request-ID") if response is not None else None,
+    )
+
+
+def _check_response(response: httpx.Response, *, executing: bool) -> None:
+    if response.status_code == 200:
+        return
+    try:
+        body = McpErrorBody.model_validate(response.json())
+    except ValueError:
+        if executing:
+            raise _unknown_response(response) from None
+        raise MCPError(
+            "MCP discovery failed without a typed gateway response",
+            execution_state="not_started",
+            status_code=response.status_code,
+            request_id=response.headers.get("X-Otari-Request-ID"),
+        ) from None
+    error_type = MCPOutcomeUnknownError if body.execution_state == "outcome_unknown" else MCPError
+    raise error_type(
+        body.detail,
+        code=body.code,
+        execution_state="outcome_unknown" if body.execution_state == "outcome_unknown" else "not_started",
+        request_id=body.request_id,
+        status_code=response.status_code,
+        retry_after=response.headers.get("Retry-After"),
+    )
+
+
+def _execution_body(
+    mcp_server_id: str | UUID,
+    tool_name: str,
+    arguments: dict[str, Any],
+    server_revision: str,
+    client_execution_id: str | UUID,
+) -> dict[str, Any]:
+    request = build_request(McpExecuteRequest, {
+        "mcp_server_id": mcp_server_id,
+        "tool_name": tool_name,
+        "arguments": arguments,
+        "server_revision": server_revision,
+        "client_execution_id": client_execution_id,
+    })
+    # JSON mode serializes UUIDs; do not exclude None recursively, since null
+    # values inside authorized arguments must survive unchanged.
+    return request.model_dump(mode="json", by_alias=True, exclude={"additional_properties"})
+
+
+def _execution_result(response: httpx.Response) -> CallToolResult:
+    _check_response(response, executing=True)
+    try:
+        payload = response.json()
+        if not isinstance(payload, dict):
+            raise TypeError("Expected an MCP result object")  # noqa: TRY301
+        return build_request(CallToolResult, payload)
+    except (ValueError, TypeError, KeyError):
+        # A malformed success cannot establish that execution did not happen.
+        raise _unknown_response(response) from None
+
+
+class MCP:
+    """Stored-server MCP discovery and caller-authorized execution."""
+
+    def __init__(self, client: OtariClient) -> None:
+        self._client = client
+
+    def list_tools(self, mcp_server_id: str | UUID) -> McpToolsResponse:
+        """Discover the authorized catalog and the stored-server revision.
+
+        Tool annotations are untrusted metadata, not authorization decisions.
+        """
+        server_id = UUID(str(mcp_server_id))
+        try:
+            response = self._client._http.get(
+                f"{self._client._base_url}/mcp/servers/{server_id}/tools",
+                headers=self._client._default_headers,
+                follow_redirects=False,
+            )
+        except httpx.RequestError:
+            raise MCPError("MCP discovery transport failed", execution_state="not_started") from None
+        _check_response(response, executing=False)
+        return McpToolsResponse.model_validate(response.json())
+
+    def execute(
+        self,
+        *,
+        mcp_server_id: str | UUID,
+        tool_name: str,
+        arguments: dict[str, Any],
+        server_revision: str,
+        client_execution_id: str | UUID,
+    ) -> CallToolResult:
+        """Execute exactly the caller-authorized call with one HTTP attempt.
+
+        The application owns approval, edited arguments, rejection and
+        cancellation. ``client_execution_id`` is correlation, NOT idempotency.
+        Persist discovery's ``server_revision`` with the authorized call.
+        Network failures and untyped responses are outcome-unknown; never
+        retry automatically or fall back to direct MCP execution.
+        A native result with ``is_error=True`` is a definitive result.
+        """
+        body = _execution_body(mcp_server_id, tool_name, arguments, server_revision, client_execution_id)
+        try:
+            response = self._client._http.post(
+                f"{self._client._base_url}/mcp/execute",
+                json=body,
+                headers=self._client._default_headers,
+                follow_redirects=False,
+            )
+        except httpx.RequestError:
+            raise _unknown_response() from None
+        return _execution_result(response)
+
+
+class AsyncMCP:
+    """Async counterpart of :class:`MCP`, with the same execution safety contract."""
+
+    def __init__(self, client: AsyncOtariClient) -> None:
+        self._client = client
+
+    async def list_tools(self, mcp_server_id: str | UUID) -> McpToolsResponse:
+        """Discover the authorized catalog and stored-server revision."""
+        server_id = UUID(str(mcp_server_id))
+        try:
+            response = await self._client._http.get(
+                f"{self._client._base_url}/mcp/servers/{server_id}/tools",
+                headers=self._client._default_headers,
+                follow_redirects=False,
+            )
+        except httpx.RequestError:
+            raise MCPError("MCP discovery transport failed", execution_state="not_started") from None
+        _check_response(response, executing=False)
+        return McpToolsResponse.model_validate(response.json())
+
+    async def execute(
+        self,
+        *,
+        mcp_server_id: str | UUID,
+        tool_name: str,
+        arguments: dict[str, Any],
+        server_revision: str,
+        client_execution_id: str | UUID,
+    ) -> CallToolResult:
+        """Execute one authorized call, without retries; see :meth:`MCP.execute`.
+
+        Task cancellation propagates normally. Cancellation after starting the
+        request does not establish whether the remote tool ran.
+        """
+        body = _execution_body(mcp_server_id, tool_name, arguments, server_revision, client_execution_id)
+        try:
+            response = await self._client._http.post(
+                f"{self._client._base_url}/mcp/execute",
+                json=body,
+                headers=self._client._default_headers,
+                follow_redirects=False,
+            )
+        except httpx.RequestError:
+            raise _unknown_response() from None
+        return _execution_result(response)

src/otari/types.py

@@ -14,11 +14,18 @@
 # Re-export the generated models that callers interact with directly.
 # Explicit ``as`` aliases make these public re-exports per PEP 484.
 # ---------------------------------------------------------------------------
+from otari._client.models.call_tool_result import CallToolResult as CallToolResult  # noqa: PLC0414
 from otari._client.models.chat_completion import ChatCompletion as ChatCompletion  # noqa: PLC0414
 from otari._client.models.chat_completion_chunk import ChatCompletionChunk as ChatCompletionChunk  # noqa: PLC0414
 from otari._client.models.create_embedding_response import (
     CreateEmbeddingResponse as CreateEmbeddingResponse,  # noqa: PLC0414
 )
+from otari._client.models.execution_state import ExecutionState as ExecutionState  # noqa: PLC0414
+from otari._client.models.mcp_error_body import McpErrorBody as McpErrorBody  # noqa: PLC0414
+from otari._client.models.mcp_execute_request import McpExecuteRequest as McpExecuteRequest  # noqa: PLC0414
+from otari._client.models.mcp_tool_definition import McpToolDefinition as McpToolDefinition  # noqa: PLC0414
+from otari._client.models.mcp_tool_warning import McpToolWarning as McpToolWarning  # noqa: PLC0414
+from otari._client.models.mcp_tools_response import McpToolsResponse as McpToolsResponse  # noqa: PLC0414
 from otari._client.models.message_response import MessageResponse as MessageResponse  # noqa: PLC0414
 from otari._client.models.model_object import ModelObject as ModelObject  # noqa: PLC0414
 from otari._client.models.moderation_response import ModerationResponse as ModerationResponse  # noqa: PLC0414