Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions scripts/generate_resources.py
Original file line number Diff line number Diff line change
Expand Up @@ -502,10 +502,16 @@ def update_client_file(client_path: Path, all_resources: list[str]) -> None:
"""
content = client_path.read_text()

# Build new import block, sorted by class name to match ruff isort
# Build new import block, sorted by class name to match ruff isort. Ruff
# sorts case-insensitively (isort case-sensitive defaults to false), so a
# plain sorted() drifts as soon as two names differ only by case position,
# e.g. AdsResource vs AdTargetingResource.
class_names = sorted(
"".join(word.title() for word in r.split("_")) + "Resource"
for r in all_resources
(
"".join(word.title() for word in r.split("_")) + "Resource"
for r in all_resources
),
key=str.lower,
)

import_block = "from ..resources import (\n"
Expand Down
14 changes: 14 additions & 0 deletions scripts/smoketest_streamable_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,20 @@ async def run_streamable_http_checks() -> None:
assert any(n.startswith("posts_") for n in names), names
print(f"[ok] list_tools -> {len(names)} tools (sample: {names[:5]})")

# fastmcp builds the synthetic search/call tools without
# annotations; clients that gate approval on hints then auto-deny
# them (codex-cli in non-interactive runs). Assert over the wire,
# not just in-process, so a serialization regression is caught too.
by_name = {t.name: t for t in tools.tools}
for synthetic in ("search_tools", "call_tool"):
assert synthetic in by_name, f"{synthetic} missing from tools/list"
assert by_name[synthetic].annotations is not None, (
f"{synthetic} has no annotations - approval-gating clients will deny it"
)
assert by_name["search_tools"].annotations.readOnlyHint is True
assert by_name["call_tool"].annotations.destructiveHint is True
print("[ok] synthetic search/call tools carry annotations")

# The real ContextVar test: invoke a tool and observe the error.
# - If plumbing is BROKEN: tool errors with "API key is required"
# (the ValueError raised by _get_client when ContextVar is unset).
Expand Down
4 changes: 2 additions & 2 deletions src/late/client/late_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,16 +17,16 @@
AdCampaignsResource,
AdCreativesResource,
AdInsightsResource,
AdTargetingResource,
AdsResource,
AdTargetingResource,
AnalyticsResource,
ApiKeysResource,
BroadcastsResource,
CallsResource,
CommentAutomationsResource,
CommentsResource,
ConnectResource,
ConnectedAppsResource,
ConnectResource,
ContactsResource,
ConversionsResource,
CustomFieldsResource,
Expand Down
50 changes: 48 additions & 2 deletions src/late/mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
import re
from contextvars import ContextVar
from datetime import datetime, timedelta
from typing import Any
from typing import TYPE_CHECKING, Any

import httpx
from fastmcp import FastMCP
Expand All @@ -40,6 +40,9 @@
from .auth import build_auth_provider
from .tool_definitions import TOOL_DEFINITIONS

if TYPE_CHECKING:
from fastmcp.tools import Tool

# Context variable to store the Zernio API key for the current connection
_zernio_api_key: ContextVar[str | None] = ContextVar("zernio_api_key", default=None)

Expand Down Expand Up @@ -88,6 +91,49 @@
"usage_get_usage",
]


class _AnnotatedBM25SearchTransform(BM25SearchTransform):
"""BM25SearchTransform whose synthetic tools carry ToolAnnotations.

fastmcp builds search_tools/call_tool without annotations (still true in
4.0.0b1). Clients that gate approval on hints treat an unannotated tool as
needing confirmation, so codex-cli auto-denies both in non-interactive
`codex exec` runs and reports "user cancelled MCP tool call".

Wrapping the parent's Tool via model_copy instead of rebuilding it keeps
the upstream closure by reference, so future upstream fixes to it (4.0
adds a catalog-membership check inside call_tool) are inherited.
"""

def _make_search_tool(self) -> Tool:
# Pure in-memory catalog search: no external calls, no state change.
return super()._make_search_tool().model_copy(
update={
"annotations": ToolAnnotations(
title="Search available tools",
readOnlyHint=True,
destructiveHint=False,
openWorldHint=False,
)
}
)

def _make_call_tool(self) -> Tool:
# Dispatches to any hidden tool, including writes to external
# platforms, so it must not claim to be read-only: that would bypass
# client approval for every write reachable through the proxy.
return super()._make_call_tool().model_copy(
update={
"annotations": ToolAnnotations(
title="Call a tool discovered via search",
readOnlyHint=False,
destructiveHint=True,
openWorldHint=True,
)
}
)


# Initialize MCP server
mcp = FastMCP(
"Zernio",
Expand Down Expand Up @@ -120,7 +166,7 @@
auth=build_auth_provider(),
# Collapse the ~383 generated tools behind search_tools/call_tool; the
# pinned ergonomic tools remain always-visible.
transforms=[BM25SearchTransform(always_visible=_PINNED_TOOLS, max_results=8)],
transforms=[_AnnotatedBM25SearchTransform(always_visible=_PINNED_TOOLS, max_results=8)],
)


Expand Down
39 changes: 39 additions & 0 deletions tests/test_search_transform_annotations.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
"""The BM25 synthetic meta-tools must carry ToolAnnotations.

fastmcp builds search_tools/call_tool without annotations (still true in
3.4.4, 3.4.5 and 4.0.0b1). Clients that gate approval on hints then treat
them as needing confirmation: codex-cli auto-denies unannotated tools in
non-interactive `codex exec` runs and reports "user cancelled MCP tool call".

These assertions are also the canary for the private-method override in
_AnnotatedBM25SearchTransform: if a fastmcp upgrade renames _make_search_tool
or _make_call_tool, the overrides become dead code and this test fails loudly
instead of the annotations silently disappearing from tools/list.
"""

from late.mcp.server import mcp


async def _synthetic_tools():
(transform,) = mcp.transforms
return {t.name: t for t in await transform.transform_tools([])}


async def test_search_tools_is_annotated_read_only():
annotations = (await _synthetic_tools())["search_tools"].annotations

assert annotations is not None
assert annotations.readOnlyHint is True
assert annotations.destructiveHint is False
assert annotations.openWorldHint is False


async def test_call_tool_is_annotated_destructive():
# The proxy can dispatch to any hidden tool, including writes to external
# platforms, so it must not claim to be read-only.
annotations = (await _synthetic_tools())["call_tool"].annotations

assert annotations is not None
assert annotations.readOnlyHint is False
assert annotations.destructiveHint is True
assert annotations.openWorldHint is True
Loading