From 7d210331baa1b17a7c526332fd46a988c8b801be Mon Sep 17 00:00:00 2001 From: Hareesh Date: Wed, 23 Sep 2026 14:43:11 +0200 Subject: [PATCH 1/3] feat: add caller-controlled MCP methods Expose sync and async stored-server discovery and execution with public generated types, required server revisions, and MCP-specific typed errors. Use HTTPX's zero-retry transport instead of the generated urllib3 retry path. Preserve indeterminate outcomes and disable redirects; test request counts beneath the connection retry loop. Document approval boundaries, correlation-only execution IDs, cancellation, and the separate gateway endpoint-manifest follow-up. Fixes #65 --- README.md | 64 ++++++++ src/otari/__init__.py | 18 +++ src/otari/async_client.py | 10 +- src/otari/client.py | 13 +- src/otari/errors.py | 32 ++++ src/otari/mcp.py | 189 ++++++++++++++++++++++++ src/otari/types.py | 7 + tests/unit/test_mcp.py | 301 ++++++++++++++++++++++++++++++++++++++ 8 files changed, 630 insertions(+), 4 deletions(-) create mode 100644 src/otari/mcp.py create mode 100644 tests/unit/test_mcp.py diff --git a/README.md b/README.md index f2e7a429..6889652b 100644 --- a/README.md +++ b/README.md @@ -324,6 +324,70 @@ for item in results.results: client.cancel_batch(batch.id, provider="openai") ``` +### Caller-controlled MCP + +Use `client.mcp.list_tools(mcp_server_id)` to discover a stored server's authorized +catalog, then `client.mcp.execute(...)` to execute one exact caller-authorized call. +Both platform tokens and self-hosted API keys work. These methods use +`GET /api/v1/mcp/servers/{mcp_server_id}/tools` and `POST /api/v1/mcp/execute`. + +```python +from uuid import uuid4 + +from otari import OtariClient + +with OtariClient(platform_token="tk_your_api_token", timeout=60) as client: + catalog = client.mcp.list_tools("2c948a61-dc96-4cd8-96bb-8e1434bf424e") + # Your application obtains authorization before executing. Persist the + # server id, revision, tool name, and final (possibly edited) arguments + # together. Rejected or cancelled proposals must not reach execute(). + result = client.mcp.execute( + mcp_server_id=catalog.server_id, + server_revision=catalog.server_revision, + tool_name="create_issue", + arguments={"title": "Caller-authorized title"}, + client_execution_id=uuid4(), + ) + print(result.is_error, result.structured_content) +``` + +With `AsyncOtariClient`, use `await client.mcp.list_tools(...)` and +`await client.mcp.execute(...)` with the same arguments. `timeout` applies to +both MCP methods, and closing the client closes their transport. + +Discovery returns `McpToolsResponse`, including `McpToolDefinition` entries and +`McpToolWarning` entries. Execution returns the generated `CallToolResult`, with +native content, `_meta` (`meta`), `structuredContent` (`structured_content`), and +`isError` (`is_error`). A result with `is_error=True` is a definitive native MCP +result, not a transport failure or retry signal. These types, `McpExecuteRequest`, +`McpErrorBody`, and `ExecutionState` are public imports from `otari` and `otari.types`. + +**Execution safety:** + +- Otari enforces server access and policy but does not obtain or verify human + approval. The application owns approval, rejection, argument editing, and + cancellation. Tool descriptions and annotations are untrusted metadata. +- `server_revision` is required. Persist discovery's revision with the authorized + call. A changed stored configuration produces `409 mcp_server_changed` with + `execution_state="not_started"`. It does not pin a remote tool's implementation. +- Execution makes one HTTP attempt after local validation, with no connection, + status, or redirect retries. `client_execution_id` is correlation only, **not + an idempotency key**. Repeating it may execute the tool again. Disable retries + in application policies, proxies, ingress controllers, and service meshes too. +- MCP failures raise `MCPError` (an `OtariError`) in both auth modes, preserving + `code`, `execution_state`, `request_id`, `status_code`, and optional `retry_after`. + `not_started` means Otari knows dispatch did not begin; the SDK still never retries. +- `MCPOutcomeUnknownError` is an `MCPError` with `execution_state="outcome_unknown"`: + the remote tool may already have run. This includes typed gateway errors, timeout + and network failures, and malformed or untyped execution responses. Without a + typed response, `code` is `None` and `request_id` is available only if a response + header supplied it. Surface an indeterminate outcome; never automatically retry + or fall back to direct MCP. Async task cancellation propagates normally and also + leaves the outcome indeterminate once the request has started. + +See the [gateway MCP contract](https://github.com/mozilla-ai/otari/blob/main/docs/mcp.md#caller-orchestrated-mcp) +for policy boundaries and execution states. + ### Error handling In platform mode, HTTP errors are mapped to typed exceptions: diff --git a/src/otari/__init__.py b/src/otari/__init__.py index 515f9a08..a30687ed 100644 --- a/src/otari/__init__.py +++ b/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", diff --git a/src/otari/async_client.py b/src/otari/async_client.py index 0b511fcb..f215f392 100644 --- a/src/otari/async_client.py +++ b/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). diff --git a/src/otari/client.py b/src/otari/client.py index 5d9f735f..4415283c 100644 --- a/src/otari/client.py +++ b/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). diff --git a/src/otari/errors.py b/src/otari/errors.py index d5adca33..28e7de8f 100644 --- a/src/otari/errors.py +++ b/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).""" diff --git a/src/otari/mcp.py b/src/otari/mcp.py new file mode 100644 index 00000000..0ff8613c --- /dev/null +++ b/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) diff --git a/src/otari/types.py b/src/otari/types.py index 722fd796..d22b1b60 100644 --- a/src/otari/types.py +++ b/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 diff --git a/tests/unit/test_mcp.py b/tests/unit/test_mcp.py new file mode 100644 index 00000000..5273a33d --- /dev/null +++ b/tests/unit/test_mcp.py @@ -0,0 +1,301 @@ +"""Offline MCP contract and single-attempt transport regression tests.""" + +from __future__ import annotations + +import asyncio +import json +from unittest.mock import AsyncMock, Mock +from uuid import UUID + +import httpcore +import httpx +import pytest +import respx + +from otari import ( + AsyncOtariClient, + CallToolResult, + MCPError, + MCPOutcomeUnknownError, + McpToolsResponse, + OtariClient, +) + +BASE = "https://gateway.example.test" +SERVER = "2c948a61-dc96-4cd8-96bb-8e1434bf424e" +EXECUTION = "6e51b3bc-6f48-4c68-a61e-0786bc80cd67" +PARAMS = { + "mcp_server_id": SERVER, + "tool_name": "create_issue", + "arguments": {"title": "Approved edit", "nested": {"null": None, "array": [1, False, "é"]}}, + "server_revision": "revision:1", + "client_execution_id": EXECUTION, +} +CATALOG = { + "server_id": SERVER, + "server_revision": "revision:1", + "tools": [{ + "name": "create_issue", + "description": "Untrusted description", + "input_schema": {"type": "object", "vendor:keyword": {"x": 1}}, + "annotations": {"readOnlyHint": False}, + }], + "warnings": [{"tool_name": "omitted", "code": "mcp_tool_schema_unsupported"}], +} +RESULT = { + "content": [{"type": "text", "text": "Created issue #42"}], + "structuredContent": {"issue_number": 42}, + "isError": False, + "_meta": {"trace": "native"}, + "vendor": {"untouched": True}, +} + + +@pytest.fixture(params=["sync", "async"]) +async def client(request, auth_mode, monkeypatch): + for name in ( + "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy", + "OTARI_AI_TOKEN", "GATEWAY_PLATFORM_TOKEN", "GATEWAY_API_KEY", + ): + monkeypatch.delenv(name, raising=False) + kwargs = {auth_mode: "test-secret"} + cls = OtariClient if request.param == "sync" else AsyncOtariClient + instance = cls(api_base=BASE, timeout=0.5, default_headers={"App-Context": "test"}, **kwargs) + yield instance + if isinstance(instance, AsyncOtariClient): + await instance.close() + else: + instance.close() + assert instance._http.is_closed + + +@pytest.fixture(params=["api_key", "platform_token"]) +def auth_mode(request): + return request.param + + +async def execute(client, **overrides): + params = {**PARAMS, **overrides} + if isinstance(client, AsyncOtariClient): + return await client.mcp.execute(**params) + return client.mcp.execute(**params) + + +async def discover(client): + if isinstance(client, AsyncOtariClient): + return await client.mcp.list_tools(UUID(SERVER)) + return client.mcp.list_tools(UUID(SERVER)) + + +def assert_auth(request, auth_mode): + expected = "Authorization" if auth_mode == "platform_token" else "Otari-Key" + absent = "Otari-Key" if auth_mode == "platform_token" else "Authorization" + assert request.headers[expected] == "Bearer test-secret" + assert absent not in request.headers + assert request.headers["App-Context"] == "test" + + +class TestMCP: + async def test_discovery(self, client, auth_mode): + with respx.mock as router: + route = router.get(f"{BASE}/api/v1/mcp/servers/{SERVER}/tools").respond(200, json=CATALOG) + result = await discover(client) + assert isinstance(result, McpToolsResponse) + assert result.to_dict() == {**CATALOG, "server_id": UUID(SERVER)} + assert route.call_count == 1 + assert_auth(route.calls[0].request, auth_mode) + assert client.mcp is client.mcp + + @pytest.mark.parametrize("is_error", [False, True]) + async def test_native_result_and_exact_arguments(self, client, auth_mode, is_error): + native = {**RESULT, "isError": is_error} + with respx.mock as router: + route = router.post(f"{BASE}/api/v1/mcp/execute").respond(200, json=native) + result = await execute(client, mcp_server_id=UUID(SERVER), client_execution_id=UUID(EXECUTION)) + assert isinstance(result, CallToolResult) + # Generated content models materialize absent nullable metadata as None. + expected_content = [{**native["content"][0], "annotations": None, "_meta": None}] + assert result.to_dict() == {**native, "content": expected_content} + assert result.is_error is is_error + assert route.call_count == 1 + request = route.calls[0].request + assert json.loads(request.content) == PARAMS + assert_auth(request, auth_mode) + assert request.extensions["timeout"]["read"] == 0.5 + + @pytest.mark.parametrize("content", [ + {"type": "image", "data": "YWJj", "mimeType": "image/png"}, + {"type": "audio", "data": "YWJj", "mimeType": "audio/wav"}, + {"type": "resource_link", "uri": "https://example.com/a", "name": "a"}, + {"type": "resource", "resource": {"uri": "https://example.com/a", "text": "native"}}, + ]) + async def test_native_content_variants(self, client, content): + with respx.mock as router: + route = router.post(f"{BASE}/api/v1/mcp/execute").respond(200, json={"content": [content]}) + result = await execute(client, arguments={}) + actual = result.content[0].to_dict() + for key, value in content.items(): + if key == "resource": + for resource_key, resource_value in value.items(): + assert actual[key][resource_key] == resource_value + else: + assert actual[key] == value + assert route.call_count == 1 + assert json.loads(route.calls[0].request.content)["arguments"] == {} + + @pytest.mark.parametrize(("status", "code", "state"), [ + (401, "authentication_failed", "not_started"), + (403, "mcp_tool_not_allowed", "not_started"), + (404, "mcp_server_not_found", "not_started"), + (409, "mcp_server_changed", "not_started"), + (422, "invalid_request", "not_started"), + (429, "rate_limit_exceeded", "not_started"), + (502, "mcp_connection_failed", "not_started"), + (503, "mcp_capacity_unavailable", "not_started"), + (504, "mcp_outcome_unknown", "outcome_unknown"), + (502, "mcp_result_too_large", "outcome_unknown"), + ]) + async def test_typed_errors(self, client, status, code, state): + body = {"detail": "Safe gateway message", "code": code, "execution_state": state, "request_id": "req_body"} + with respx.mock as router: + route = router.post(f"{BASE}/api/v1/mcp/execute").respond( + status, json=body, headers={"Retry-After": "2", "X-Otari-Request-ID": "req_header"}, + ) + with pytest.raises(MCPError) as caught: + await execute(client) + error = caught.value + assert error.code == code + assert error.execution_state == state + assert error.request_id == "req_body" + assert error.status_code == status + assert error.retry_after == "2" + assert error.message == body["detail"] + assert isinstance(error, MCPOutcomeUnknownError) == (state == "outcome_unknown") + assert route.call_count == 1 + + async def test_discovery_typed_error(self, client): + with respx.mock as router: + route = router.get(f"{BASE}/api/v1/mcp/servers/{SERVER}/tools").respond(404, json={ + "detail": "MCP server not found", "code": "mcp_server_not_found", + "execution_state": "not_started", "request_id": "req_discovery", + }) + with pytest.raises(MCPError) as caught: + await discover(client) + assert caught.value.code == "mcp_server_not_found" + assert caught.value.request_id == "req_discovery" + assert caught.value.execution_state == "not_started" + assert route.call_count == 1 + + @pytest.mark.parametrize("failure", [ + httpx.ConnectTimeout, httpx.ReadTimeout, httpx.WriteTimeout, httpx.PoolTimeout, + httpx.ConnectError, httpx.ReadError, httpx.WriteError, httpx.RemoteProtocolError, + ]) + async def test_transport_failure_is_unknown(self, client, failure, caplog): + with respx.mock as router: + route = router.post(f"{BASE}/api/v1/mcp/execute").mock(side_effect=failure("SECRET arguments")) + with pytest.raises(MCPOutcomeUnknownError) as caught: + await execute(client) + assert route.call_count == 1 + assert caught.value.execution_state == "outcome_unknown" + assert caught.value.code is None + assert caught.value.request_id is None + assert caught.value.status_code is None + assert "SECRET" not in str(caught.value) + assert "SECRET" not in caplog.text + assert caught.value.original_error is None + assert caught.value.__suppress_context__ + + @pytest.mark.parametrize(("status", "body"), [ + (502, "proxy failure SECRET"), (504, "timeout"), (200, "invalid JSON"), + (200, "{}"), (200, "null"), (200, '{"content": "invalid"}'), + (429, '{"detail":"untyped"}'), + (500, '{"execution_state":"not_started"}'), + ]) + async def test_untyped_response_is_unknown(self, client, status, body): + with respx.mock as router: + route = router.post(f"{BASE}/api/v1/mcp/execute").respond( + status, text=body, headers={"X-Otari-Request-ID": "req_header"}, + ) + with pytest.raises(MCPOutcomeUnknownError) as caught: + await execute(client) + assert route.call_count == 1 + assert caught.value.request_id == "req_header" + assert caught.value.code is None + assert "SECRET" not in str(caught.value) + + @pytest.mark.parametrize("status", [301, 302, 303, 307, 308]) + async def test_no_redirect(self, client, status): + with respx.mock as router: + route = router.post(f"{BASE}/api/v1/mcp/execute").respond( + status, headers={"Location": f"{BASE}/redirected"}, + ) + redirected = router.route(url=f"{BASE}/redirected").respond(200, json=RESULT) + with pytest.raises(MCPOutcomeUnknownError): + await execute(client) + assert route.call_count == 1 + assert redirected.call_count == 0 + + async def test_correlation_is_not_deduplication(self, client): + with respx.mock as router: + route = router.post(f"{BASE}/api/v1/mcp/execute").respond(200, json=RESULT) + await execute(client) + await execute(client) + assert route.call_count == 2 + + async def test_invalid_request_never_sent(self, client): + with respx.mock as router: + with pytest.raises(ValueError, match="server_revision"): + await execute(client, server_revision="") + assert router.calls.call_count == 0 + + @pytest.mark.parametrize("failure", [httpcore.ConnectError, httpcore.ConnectTimeout]) + async def test_actual_transport_does_not_retry_connection(self, client, failure, monkeypatch): + # Below HTTPX and httpcore's retry loop, unlike respx or a mocked .post. + # No real socket is opened. A regression enabling connection retries + # increases this counter even though the wrapper calls .post only once. + backend = client._http._transport._pool._network_backend + mock_type = AsyncMock if isinstance(client, AsyncOtariClient) else Mock + connect = mock_type(side_effect=failure("mock connection failure")) + monkeypatch.setattr(backend, "connect_tcp", connect) + with pytest.raises(MCPOutcomeUnknownError): + await execute(client) + assert connect.call_count == 1 + + async def test_environment_proxy_transport_has_no_retries(self, client, monkeypatch): + monkeypatch.setenv("HTTPS_PROXY", "http://proxy.example.test:8080") + cls = type(client) + proxied = cls(api_base=BASE, api_key="test-secret") + try: + transport = proxied._http._transport_for_url(httpx.URL(BASE)) + assert transport._pool._retries == 0 + backend = transport._pool._network_backend + mock_type = AsyncMock if isinstance(proxied, AsyncOtariClient) else Mock + connect = mock_type(side_effect=httpcore.ConnectError("mock proxy failure")) + monkeypatch.setattr(backend, "connect_tcp", connect) + with pytest.raises(MCPOutcomeUnknownError): + await execute(proxied) + assert connect.call_count == 1 + assert connect.call_args.kwargs["host"] == "proxy.example.test" + finally: + if isinstance(proxied, AsyncOtariClient): + await proxied.close() + else: + proxied.close() + + async def test_async_cancellation_does_not_retry(self, client): + if not isinstance(client, AsyncOtariClient): + return + + # respx does not record BaseException failures such as CancelledError. + # Count at the callback itself, before cancellation unwinds the stack. + attempts = Mock() + + def cancel(request): + attempts(request) + raise asyncio.CancelledError + + with respx.mock as router: + router.post(f"{BASE}/api/v1/mcp/execute").mock(side_effect=cancel) + with pytest.raises(asyncio.CancelledError): + await execute(client) + assert attempts.call_count == 1 From 03b5d84000cb31e5073af2055abe915894509c02 Mon Sep 17 00:00:00 2001 From: Hareesh Date: Wed, 23 Sep 2026 14:52:26 +0200 Subject: [PATCH 2/3] docs: revert MCP README additions --- README.md | 64 ------------------------------------------------------- 1 file changed, 64 deletions(-) diff --git a/README.md b/README.md index 6889652b..f2e7a429 100644 --- a/README.md +++ b/README.md @@ -324,70 +324,6 @@ for item in results.results: client.cancel_batch(batch.id, provider="openai") ``` -### Caller-controlled MCP - -Use `client.mcp.list_tools(mcp_server_id)` to discover a stored server's authorized -catalog, then `client.mcp.execute(...)` to execute one exact caller-authorized call. -Both platform tokens and self-hosted API keys work. These methods use -`GET /api/v1/mcp/servers/{mcp_server_id}/tools` and `POST /api/v1/mcp/execute`. - -```python -from uuid import uuid4 - -from otari import OtariClient - -with OtariClient(platform_token="tk_your_api_token", timeout=60) as client: - catalog = client.mcp.list_tools("2c948a61-dc96-4cd8-96bb-8e1434bf424e") - # Your application obtains authorization before executing. Persist the - # server id, revision, tool name, and final (possibly edited) arguments - # together. Rejected or cancelled proposals must not reach execute(). - result = client.mcp.execute( - mcp_server_id=catalog.server_id, - server_revision=catalog.server_revision, - tool_name="create_issue", - arguments={"title": "Caller-authorized title"}, - client_execution_id=uuid4(), - ) - print(result.is_error, result.structured_content) -``` - -With `AsyncOtariClient`, use `await client.mcp.list_tools(...)` and -`await client.mcp.execute(...)` with the same arguments. `timeout` applies to -both MCP methods, and closing the client closes their transport. - -Discovery returns `McpToolsResponse`, including `McpToolDefinition` entries and -`McpToolWarning` entries. Execution returns the generated `CallToolResult`, with -native content, `_meta` (`meta`), `structuredContent` (`structured_content`), and -`isError` (`is_error`). A result with `is_error=True` is a definitive native MCP -result, not a transport failure or retry signal. These types, `McpExecuteRequest`, -`McpErrorBody`, and `ExecutionState` are public imports from `otari` and `otari.types`. - -**Execution safety:** - -- Otari enforces server access and policy but does not obtain or verify human - approval. The application owns approval, rejection, argument editing, and - cancellation. Tool descriptions and annotations are untrusted metadata. -- `server_revision` is required. Persist discovery's revision with the authorized - call. A changed stored configuration produces `409 mcp_server_changed` with - `execution_state="not_started"`. It does not pin a remote tool's implementation. -- Execution makes one HTTP attempt after local validation, with no connection, - status, or redirect retries. `client_execution_id` is correlation only, **not - an idempotency key**. Repeating it may execute the tool again. Disable retries - in application policies, proxies, ingress controllers, and service meshes too. -- MCP failures raise `MCPError` (an `OtariError`) in both auth modes, preserving - `code`, `execution_state`, `request_id`, `status_code`, and optional `retry_after`. - `not_started` means Otari knows dispatch did not begin; the SDK still never retries. -- `MCPOutcomeUnknownError` is an `MCPError` with `execution_state="outcome_unknown"`: - the remote tool may already have run. This includes typed gateway errors, timeout - and network failures, and malformed or untyped execution responses. Without a - typed response, `code` is `None` and `request_id` is available only if a response - header supplied it. Surface an indeterminate outcome; never automatically retry - or fall back to direct MCP. Async task cancellation propagates normally and also - leaves the outcome indeterminate once the request has started. - -See the [gateway MCP contract](https://github.com/mozilla-ai/otari/blob/main/docs/mcp.md#caller-orchestrated-mcp) -for policy boundaries and execution states. - ### Error handling In platform mode, HTTP errors are mapped to typed exceptions: From 2a100247d6a2de63bf19e83f492a240eb82d6428 Mon Sep 17 00:00:00 2001 From: Hareesh Date: Wed, 23 Sep 2026 18:08:24 +0200 Subject: [PATCH 3/3] fix(mcp): preserve completed execution state --- src/otari/errors.py | 5 +++-- src/otari/mcp.py | 2 +- tests/unit/test_mcp.py | 1 + 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/src/otari/errors.py b/src/otari/errors.py index 28e7de8f..36d8f54d 100644 --- a/src/otari/errors.py +++ b/src/otari/errors.py @@ -48,14 +48,15 @@ class MCPError(OtariError): ``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. + tool may already have run; ``completed`` preserves a known completed + execution. No state triggers an SDK retry. """ def __init__( self, message: str, *, - execution_state: Literal["not_started", "outcome_unknown"], + execution_state: Literal["not_started", "outcome_unknown", "completed"], code: str | None = None, request_id: str | None = None, status_code: int | None = None, diff --git a/src/otari/mcp.py b/src/otari/mcp.py index 0ff8613c..ae3ccc3c 100644 --- a/src/otari/mcp.py +++ b/src/otari/mcp.py @@ -50,7 +50,7 @@ def _check_response(response: httpx.Response, *, executing: bool) -> None: raise error_type( body.detail, code=body.code, - execution_state="outcome_unknown" if body.execution_state == "outcome_unknown" else "not_started", + execution_state=body.execution_state.value, request_id=body.request_id, status_code=response.status_code, retry_after=response.headers.get("Retry-After"), diff --git a/tests/unit/test_mcp.py b/tests/unit/test_mcp.py index 5273a33d..0fd7d6ca 100644 --- a/tests/unit/test_mcp.py +++ b/tests/unit/test_mcp.py @@ -154,6 +154,7 @@ async def test_native_content_variants(self, client, content): (503, "mcp_capacity_unavailable", "not_started"), (504, "mcp_outcome_unknown", "outcome_unknown"), (502, "mcp_result_too_large", "outcome_unknown"), + (502, "mcp_result_too_large", "completed"), ]) async def test_typed_errors(self, client, status, code, state): body = {"detail": "Safe gateway message", "code": code, "execution_state": state, "request_id": "req_body"}