From c459155995c1a6f0cc956327515f9a2de61c2378 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Sat, 4 Jul 2026 02:53:04 -0400 Subject: [PATCH 1/2] feat: add live-model validation for MCP tool gating --- python/examples/tool_gating/README.md | 4 ++ .../tool_gating/mcp_calendar_admin/README.md | 68 +++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/python/examples/tool_gating/README.md b/python/examples/tool_gating/README.md index b7299bd..947f4c3 100644 --- a/python/examples/tool_gating/README.md +++ b/python/examples/tool_gating/README.md @@ -49,3 +49,7 @@ prohibit calendar_admin the MCP tool is absent from the exposed tool set and blocked if invoked directly. + +This example also includes an opt-in live-model comparison showing that a real +tool-calling model can only act through the MCP tool surface the host exposes +from authoritative Context Compiler state. diff --git a/python/examples/tool_gating/mcp_calendar_admin/README.md b/python/examples/tool_gating/mcp_calendar_admin/README.md index 4a5a848..79be2db 100644 --- a/python/examples/tool_gating/mcp_calendar_admin/README.md +++ b/python/examples/tool_gating/mcp_calendar_admin/README.md @@ -23,3 +23,71 @@ prohibit calendar_admin If a caller still invokes the hidden tool directly, the host blocks execution. Adversarial request text does not expose the tool or mutate policy state. + +The provider-free tests are the canonical proof for this example. + +This example also includes an opt-in live-model comparison that shows the same +admin-calendar intent reaches different outcomes because authoritative Context +Compiler state changes the model-visible tool surface. + +## Live-model comparison + +The live-model path keeps the same host-owned MCP tool registry and execution +rules. + +The host exposes only the MCP tools allowed by authoritative state, then asks a +real tool-calling model to complete the same admin action. + +The model can only use the tools the host exposes. + +This is not prompt reinjection: + +- the host does not derive authority from model output +- the model does not create or mutate authoritative state +- Context Compiler state determines whether the protected admin tool is visible + and executable + +### Same request, different state + +User intent: + +```text +Create an admin calendar event named Quarterly access review on calendar ops-admin. +``` + +Outcome matrix: + +- absent state: protected tool is not exposed, and no protected side effect + occurs +- `use calendar_admin`: protected tool is exposed; if the model selects it, the + host executes it and writes one side effect +- contradiction with `prohibit calendar_admin`: Context Compiler returns + clarify/conflict and blocks protected execution before tool execution + +### Validation + +Canonical provider-free tests: + +```bash +uv run --no-sync pytest python/tests/test_mcp_calendar_admin_tool_gating_example.py +uv run --no-sync pytest python/tests/test_mcp_calendar_admin_live_model_helper.py +``` + +Opt-in live-model validation: + +```bash +export RUN_MCP_CALENDAR_ADMIN_LIVE_MODEL=1 +export MODEL=openai/gpt-4o-mini +export OPENAI_API_KEY=... +uv run --no-sync pytest python/tests/test_mcp_calendar_admin_live_model.py +``` + +Optional compatible-provider settings: + +- `PROVIDER` +- `OPENAI_BASE_URL` + +For local Ollama, this example accepts either: + +- `PROVIDER=ollama` with a bare `MODEL` such as `qwen2.5:1.5b-instruct` +- or an explicit LiteLLM-style `MODEL=ollama/qwen2.5:1.5b-instruct` From 07b42022e300785275573cc5127f9c66699aa091 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Sat, 4 Jul 2026 03:36:07 -0400 Subject: [PATCH 2/2] feat: add live-model validation for MCP tool gating --- python/examples/_shared/__init__.py | 1 + python/examples/_shared/litellm_request.py | 61 +++ .../fastapi/model_approval.py | 8 +- .../mcp_calendar_admin/live_model.py | 368 ++++++++++++++++++ .../test_mcp_calendar_admin_live_model.py | 75 ++++ ...st_mcp_calendar_admin_live_model_helper.py | 129 ++++++ ..._mcp_calendar_admin_live_model_provider.py | 183 +++++++++ 7 files changed, 820 insertions(+), 5 deletions(-) create mode 100644 python/examples/_shared/__init__.py create mode 100644 python/examples/_shared/litellm_request.py create mode 100644 python/examples/tool_gating/mcp_calendar_admin/live_model.py create mode 100644 python/tests/test_mcp_calendar_admin_live_model.py create mode 100644 python/tests/test_mcp_calendar_admin_live_model_helper.py create mode 100644 python/tests/test_mcp_calendar_admin_live_model_provider.py diff --git a/python/examples/_shared/__init__.py b/python/examples/_shared/__init__.py new file mode 100644 index 0000000..50c940a --- /dev/null +++ b/python/examples/_shared/__init__.py @@ -0,0 +1 @@ +"""Shared helpers for Python examples.""" diff --git a/python/examples/_shared/litellm_request.py b/python/examples/_shared/litellm_request.py new file mode 100644 index 0000000..320ac14 --- /dev/null +++ b/python/examples/_shared/litellm_request.py @@ -0,0 +1,61 @@ +"""Tiny shared helper for LiteLLM live-validation request kwargs.""" + +from __future__ import annotations + +from typing import Literal, TypedDict + + +class ProviderConfigLike(TypedDict): + mode: Literal["openai", "ollama", "openai_compatible"] + source: str + base_url: str + model: str + api_key: str | None + + +class LiteLLMProviderKwargs(TypedDict, total=False): + model: str + api_base: str + api_key: str + temperature: float + drop_params: bool + + +def build_litellm_provider_kwargs( + config: ProviderConfigLike, +) -> LiteLLMProviderKwargs: + """Return LiteLLM-safe provider/model kwargs for live validations.""" + + model = _normalize_litellm_model_id(config) + base_url = _config_value(config, "base_url") + if base_url is None: + raise RuntimeError("Provider config missing base_url.") + + kwargs: LiteLLMProviderKwargs = { + "model": model, + "api_base": base_url, + "drop_params": True, + } + api_key = _config_value(config, "api_key") + if api_key: + kwargs["api_key"] = api_key + kwargs["temperature"] = 0 + return kwargs + + +def _normalize_litellm_model_id(config: ProviderConfigLike) -> str: + mode = _config_value(config, "mode") + model = _config_value(config, "model") + if model is None: + raise RuntimeError("Provider config missing model.") + if mode == "ollama" and "/" not in model: + return f"ollama/{model}" + return model + + +def _config_value(config: object, key: str) -> str | None: + if isinstance(config, dict): + value = config.get(key) + else: + value = getattr(config, key, None) + return value if isinstance(value, str) or value is None else str(value) diff --git a/python/examples/execution_authorization/expense_approval/fastapi/model_approval.py b/python/examples/execution_authorization/expense_approval/fastapi/model_approval.py index faf99e1..54aad47 100644 --- a/python/examples/execution_authorization/expense_approval/fastapi/model_approval.py +++ b/python/examples/execution_authorization/expense_approval/fastapi/model_approval.py @@ -7,6 +7,8 @@ from importlib import import_module from typing import cast +from python.examples._shared.litellm_request import build_litellm_provider_kwargs + try: from host_support import print_startup_config, resolve_provider_config except ImportError: @@ -80,13 +82,9 @@ def get_model_approval_claim(expense_summary: str) -> ModelApproval: f"{expense_summary}" ) kwargs: dict[str, object] = { - "model": config.model, + **build_litellm_provider_kwargs(config), "messages": [{"role": "user", "content": prompt}], - "temperature": 0, - "api_base": config.base_url, } - if config.api_key: - kwargs["api_key"] = config.api_key response = completion(**kwargs) message = _extract_response_content(response) diff --git a/python/examples/tool_gating/mcp_calendar_admin/live_model.py b/python/examples/tool_gating/mcp_calendar_admin/live_model.py new file mode 100644 index 0000000..8593d7a --- /dev/null +++ b/python/examples/tool_gating/mcp_calendar_admin/live_model.py @@ -0,0 +1,368 @@ +"""Opt-in live-model tool-gating comparison for MCP calendar admin.""" + +from __future__ import annotations + +import json +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from importlib import import_module +from pathlib import Path +from typing import Literal, TypedDict, cast + +from context_compiler import State, create_engine, get_decision_state, is_clarify + +from python.examples._shared.litellm_request import build_litellm_provider_kwargs +from python.examples.tool_gating.mcp_calendar_admin.example import ( + CalendarAdminMcpHost, + McpToolCall, +) + +try: + from host_support import print_startup_config, resolve_provider_config +except ImportError: + from host_support.provider_mode import print_startup_config, resolve_provider_config + + +class SideEffectRecord(TypedDict): + tool_name: str + calendar_id: str + event_title: str + authorization_source: Literal["context_compiler_state"] + + +class LiveModelResult(TypedDict): + decision_kind: Literal["clarify", "update", "passthrough"] | None + prompt_to_user: str | None + exposed_tool_names: list[str] + hidden_tool_names: list[str] + protected_tool_exposed: bool + model_selected_tool_name: str | None + executed: bool + blocked_reason: str | None + tool_result: str | None + execution_log: list[str] + side_effect_path: str + side_effect_count: int + + +class _LiteLLMCallKwargs(TypedDict, total=False): + model: str + messages: list[dict[str, str]] + tools: list[dict[str, object]] + tool_choice: str + temperature: float + drop_params: bool + api_base: str + api_key: str + + +@dataclass(frozen=True) +class _SelectedToolCall: + name: str | None + arguments: dict[str, str] + + +@dataclass +class CalendarAdminSideEffectStore: + """Host-owned append-only artifact for successful protected execution.""" + + artifact_path: Path + + def append(self, *, tool_call: McpToolCall) -> None: + record: SideEffectRecord = { + "tool_name": tool_call["tool_name"], + "calendar_id": tool_call["arguments"]["calendar_id"], + "event_title": tool_call["arguments"]["event_title"], + "authorization_source": "context_compiler_state", + } + self.artifact_path.parent.mkdir(parents=True, exist_ok=True) + with self.artifact_path.open("a", encoding="utf-8") as artifact: + artifact.write(json.dumps(record, sort_keys=True) + "\n") + + def count(self) -> int: + if not self.artifact_path.exists(): + return 0 + with self.artifact_path.open(encoding="utf-8") as artifact: + return sum(1 for _ in artifact) + + +def _get_litellm_completion() -> Callable[..., object]: + litellm_module = import_module("litellm") + return cast(Callable[..., object], litellm_module.completion) + + +def _state_for_request(authoritative_state: State | None) -> State | None: + return authoritative_state + + +def _decision_kind_name( + decision: object, +) -> Literal["clarify", "update", "passthrough"]: + if not isinstance(decision, dict): + raise ValueError("unexpected decision shape") + + kind = decision.get("kind") + kind_name = getattr(kind, "value", None) + if kind_name not in {"clarify", "update", "passthrough"}: + raise ValueError(f"unexpected decision kind: {kind_name}") + + return kind_name + + +def _exposed_tool_names( + host: CalendarAdminMcpHost, state: State +) -> tuple[list[str], list[str]]: + exposed_tools = host.exposed_mcp_tools(state) + return ( + [tool["name"] for tool in exposed_tools["tools"]], + exposed_tools["hidden_tool_names"], + ) + + +def _build_openai_tools( + host: CalendarAdminMcpHost, state: State +) -> list[dict[str, object]]: + exposed_tools = host.exposed_mcp_tools(state)["tools"] + tools: list[dict[str, object]] = [] + for tool in exposed_tools: + if tool["name"] == "calendar_view_events": + input_schema: dict[str, object] = { + "type": "object", + "properties": { + "calendar_id": { + "type": "string", + "description": "Calendar identifier to inspect.", + } + }, + "required": ["calendar_id"], + "additionalProperties": False, + } + elif tool["name"] == "calendar_admin_create_event": + input_schema = { + "type": "object", + "properties": { + "calendar_id": { + "type": "string", + "description": "Administrative calendar identifier.", + }, + "event_title": { + "type": "string", + "description": "Administrative event title to create.", + }, + }, + "required": ["calendar_id", "event_title"], + "additionalProperties": False, + } + else: + continue + + tools.append( + { + "type": "function", + "function": { + "name": tool["name"], + "description": tool["description"], + "parameters": input_schema, + }, + } + ) + return tools + + +def _build_messages(user_intent: str) -> list[dict[str, str]]: + return [ + { + "role": "system", + "content": ( + "You are a careful assistant.\n" + "Use a tool if one is available to complete the user's request.\n" + "Do not invent unavailable tools." + ), + }, + {"role": "user", "content": user_intent}, + ] + + +def _mapping_value(obj: object, key: str) -> object | None: + if isinstance(obj, Mapping): + return obj.get(key) + return getattr(obj, key, None) + + +def _extract_selected_tool_call(response: object) -> _SelectedToolCall: + choices = _mapping_value(response, "choices") + if not isinstance(choices, list) or not choices: + choices = getattr(response, "choices", None) + if not isinstance(choices, list) or not choices: + return _SelectedToolCall(name=None, arguments={}) + + message = _mapping_value(choices[0], "message") + if message is None: + return _SelectedToolCall(name=None, arguments={}) + + tool_calls = _mapping_value(message, "tool_calls") + if not isinstance(tool_calls, list) or not tool_calls: + return _SelectedToolCall(name=None, arguments={}) + + first_tool_call = tool_calls[0] + function_payload = _mapping_value(first_tool_call, "function") + if function_payload is None: + return _SelectedToolCall(name=None, arguments={}) + + name = _mapping_value(function_payload, "name") + raw_arguments = _mapping_value(function_payload, "arguments") + if not isinstance(name, str): + return _SelectedToolCall(name=None, arguments={}) + + if isinstance(raw_arguments, str): + decoded_arguments = json.loads(raw_arguments) + elif isinstance(raw_arguments, Mapping): + decoded_arguments = dict(raw_arguments) + else: + decoded_arguments = {} + + arguments = { + key: str(value) + for key, value in decoded_arguments.items() + if isinstance(key, str) + } + return _SelectedToolCall(name=name, arguments=arguments) + + +def _call_live_model( + *, user_intent: str, tools: list[dict[str, object]] +) -> _SelectedToolCall: + try: + completion = _get_litellm_completion() + except ModuleNotFoundError as exc: + raise RuntimeError( + "litellm is required. Install with: pip install litellm" + ) from exc + + config = resolve_provider_config(default_model="openai/gpt-4o-mini") + print_startup_config(config) + + kwargs: _LiteLLMCallKwargs = { + **build_litellm_provider_kwargs(config), + "messages": _build_messages(user_intent), + "tools": tools, + "tool_choice": "auto", + } + + response = completion(**kwargs) + return _extract_selected_tool_call(response) + + +def run_live_model_turn( + *, + user_intent: str, + authoritative_state: State | None = None, + compiler_input: str = "", + artifact_path: Path | None = None, + model_tool_selector: Callable[..., _SelectedToolCall] | None = None, +) -> LiveModelResult: + """Run one tool-gated turn with a live model or injected selector.""" + + if artifact_path is None: + artifact_path = Path( + "/tmp/context_compiler_mcp_calendar_admin/tool_calls.jsonl" + ) + + side_effect_store = CalendarAdminSideEffectStore(artifact_path=artifact_path) + host = CalendarAdminMcpHost() + engine = create_engine(state=_state_for_request(authoritative_state)) + decision_kind: Literal["clarify", "update", "passthrough"] | None = None + prompt_to_user: str | None = None + effective_state = engine.state + + if compiler_input: + decision = engine.step(compiler_input) + decision_kind = _decision_kind_name(decision) + prompt_to_user = decision.get("prompt_to_user") + if is_clarify(decision): + exposed_tool_names, hidden_tool_names = _exposed_tool_names( + host, engine.state + ) + return { + "decision_kind": decision_kind, + "prompt_to_user": prompt_to_user, + "exposed_tool_names": exposed_tool_names, + "hidden_tool_names": hidden_tool_names, + "protected_tool_exposed": "calendar_admin_create_event" + in exposed_tool_names, + "model_selected_tool_name": None, + "executed": False, + "blocked_reason": ( + "clarification required before exposing calendar admin MCP tools" + ), + "tool_result": None, + "execution_log": host.execution_log.copy(), + "side_effect_path": str(side_effect_store.artifact_path), + "side_effect_count": side_effect_store.count(), + } + + decision_state = get_decision_state(decision) + effective_state = decision_state if decision_state is not None else engine.state + + exposed_tool_names, hidden_tool_names = _exposed_tool_names(host, effective_state) + protected_tool_exposed = "calendar_admin_create_event" in exposed_tool_names + tools = _build_openai_tools(host, effective_state) + selector = model_tool_selector or _call_live_model + selected_tool_call = selector(user_intent=user_intent, tools=tools) + + if selected_tool_call.name != "calendar_admin_create_event": + return { + "decision_kind": decision_kind, + "prompt_to_user": prompt_to_user, + "exposed_tool_names": exposed_tool_names, + "hidden_tool_names": hidden_tool_names, + "protected_tool_exposed": protected_tool_exposed, + "model_selected_tool_name": selected_tool_call.name, + "executed": False, + "blocked_reason": ( + "model did not select protected admin tool" + if protected_tool_exposed + else "protected admin tool was not exposed to the model" + ), + "tool_result": None, + "execution_log": host.execution_log.copy(), + "side_effect_path": str(side_effect_store.artifact_path), + "side_effect_count": side_effect_store.count(), + } + + if not protected_tool_exposed: + return { + "decision_kind": decision_kind, + "prompt_to_user": prompt_to_user, + "exposed_tool_names": exposed_tool_names, + "hidden_tool_names": hidden_tool_names, + "protected_tool_exposed": False, + "model_selected_tool_name": selected_tool_call.name, + "executed": False, + "blocked_reason": "calendar_admin state not authorized", + "tool_result": None, + "execution_log": host.execution_log.copy(), + "side_effect_path": str(side_effect_store.artifact_path), + "side_effect_count": side_effect_store.count(), + } + + tool_call: McpToolCall = { + "tool_name": "calendar_admin_create_event", + "arguments": selected_tool_call.arguments, + } + tool_result = host.execute_mcp_tool(tool_call) + side_effect_store.append(tool_call=tool_call) + return { + "decision_kind": decision_kind, + "prompt_to_user": prompt_to_user, + "exposed_tool_names": exposed_tool_names, + "hidden_tool_names": hidden_tool_names, + "protected_tool_exposed": True, + "model_selected_tool_name": selected_tool_call.name, + "executed": True, + "blocked_reason": None, + "tool_result": tool_result, + "execution_log": host.execution_log.copy(), + "side_effect_path": str(side_effect_store.artifact_path), + "side_effect_count": side_effect_store.count(), + } diff --git a/python/tests/test_mcp_calendar_admin_live_model.py b/python/tests/test_mcp_calendar_admin_live_model.py new file mode 100644 index 0000000..3418cfc --- /dev/null +++ b/python/tests/test_mcp_calendar_admin_live_model.py @@ -0,0 +1,75 @@ +import os +from pathlib import Path + +import pytest +from context_compiler import create_engine + +from python.examples.tool_gating.mcp_calendar_admin.live_model import ( + run_live_model_turn, +) + +RUN_LIVE_MODEL_ENV_VAR = "RUN_MCP_CALENDAR_ADMIN_LIVE_MODEL" +USER_INTENT = ( + "Create an admin calendar event named Quarterly access review on calendar " + "ops-admin." +) + +pytestmark = pytest.mark.skipif( + os.getenv(RUN_LIVE_MODEL_ENV_VAR) != "1", + reason=( + "Opt-in MCP calendar-admin live-model validation. " + f"Set {RUN_LIVE_MODEL_ENV_VAR}=1 to run." + ), +) + + +def _read_jsonl(path: Path) -> list[str]: + if not path.exists(): + return [] + return path.read_text(encoding="utf-8").splitlines() + + +def test_live_model_tool_surface_changes_with_authoritative_state( + tmp_path: Path, +) -> None: + artifact_path = tmp_path / "tool_calls.jsonl" + + absent_result = run_live_model_turn( + user_intent=USER_INTENT, + artifact_path=artifact_path, + ) + + assert absent_result["protected_tool_exposed"] is False + assert absent_result["executed"] is False + assert _read_jsonl(artifact_path) == [] + + allowed_engine = create_engine() + allowed_engine.step("use calendar_admin") + allowed_result = run_live_model_turn( + user_intent=USER_INTENT, + authoritative_state=allowed_engine.state, + artifact_path=artifact_path, + ) + + assert allowed_result["protected_tool_exposed"] is True + if allowed_result["model_selected_tool_name"] != "calendar_admin_create_event": + raise AssertionError( + "Protected tool was exposed but the live model did not select " + "`calendar_admin_create_event`. " + f"Selected tool: {allowed_result['model_selected_tool_name']!r}. " + "This opt-in validation requires the model to exercise the protected " + "tool path explicitly." + ) + assert allowed_result["executed"] is True + assert len(_read_jsonl(artifact_path)) == 1 + + clarify_result = run_live_model_turn( + user_intent=USER_INTENT, + authoritative_state=allowed_engine.state, + compiler_input="prohibit calendar_admin", + artifact_path=artifact_path, + ) + + assert clarify_result["decision_kind"] == "clarify" + assert clarify_result["executed"] is False + assert len(_read_jsonl(artifact_path)) == 1 diff --git a/python/tests/test_mcp_calendar_admin_live_model_helper.py b/python/tests/test_mcp_calendar_admin_live_model_helper.py new file mode 100644 index 0000000..5911568 --- /dev/null +++ b/python/tests/test_mcp_calendar_admin_live_model_helper.py @@ -0,0 +1,129 @@ +from pathlib import Path + +from context_compiler import create_engine + +from python.examples.tool_gating.mcp_calendar_admin.live_model import ( + _SelectedToolCall, + run_live_model_turn, +) + + +def _read_jsonl(path: Path) -> list[str]: + if not path.exists(): + return [] + return path.read_text(encoding="utf-8").splitlines() + + +def test_absent_state_hides_protected_tool_from_model_and_writes_none( + tmp_path: Path, +) -> None: + artifact_path = tmp_path / "tool_calls.jsonl" + + result = run_live_model_turn( + user_intent=( + "Create an admin calendar event named Quarterly access review on " + "calendar ops-admin." + ), + artifact_path=artifact_path, + model_tool_selector=lambda **_: _SelectedToolCall( + name="calendar_view_events", + arguments={"calendar_id": "ops-admin"}, + ), + ) + + assert result["protected_tool_exposed"] is False + assert "calendar_admin_create_event" not in result["exposed_tool_names"] + assert result["executed"] is False + assert result["side_effect_count"] == 0 + assert _read_jsonl(artifact_path) == [] + + +def test_authorized_state_exposes_protected_tool_and_executes_selected_tool( + tmp_path: Path, +) -> None: + artifact_path = tmp_path / "tool_calls.jsonl" + engine = create_engine() + engine.step("use calendar_admin") + + result = run_live_model_turn( + user_intent=( + "Create an admin calendar event named Quarterly access review on " + "calendar ops-admin." + ), + authoritative_state=engine.state, + artifact_path=artifact_path, + model_tool_selector=lambda **_: _SelectedToolCall( + name="calendar_admin_create_event", + arguments={ + "calendar_id": "ops-admin", + "event_title": "Quarterly access review", + }, + ), + ) + + assert result["protected_tool_exposed"] is True + assert "calendar_admin_create_event" in result["exposed_tool_names"] + assert result["model_selected_tool_name"] == "calendar_admin_create_event" + assert result["executed"] is True + assert result["tool_result"] == ( + "created event 'Quarterly access review' on calendar 'ops-admin'" + ) + assert result["side_effect_count"] == 1 + records = _read_jsonl(artifact_path) + assert len(records) == 1 + assert '"authorization_source": "context_compiler_state"' in records[0] + + +def test_authorized_state_reports_clear_diagnostic_when_model_skips_protected_tool( + tmp_path: Path, +) -> None: + artifact_path = tmp_path / "tool_calls.jsonl" + engine = create_engine() + engine.step("use calendar_admin") + + result = run_live_model_turn( + user_intent=( + "Create an admin calendar event named Quarterly access review on " + "calendar ops-admin." + ), + authoritative_state=engine.state, + artifact_path=artifact_path, + model_tool_selector=lambda **_: _SelectedToolCall( + name="calendar_view_events", + arguments={"calendar_id": "ops-admin"}, + ), + ) + + assert result["protected_tool_exposed"] is True + assert result["executed"] is False + assert result["blocked_reason"] == "model did not select protected admin tool" + assert _read_jsonl(artifact_path) == [] + + +def test_contradiction_blocks_before_model_tool_execution(tmp_path: Path) -> None: + artifact_path = tmp_path / "tool_calls.jsonl" + engine = create_engine() + engine.step("use calendar_admin") + model_called = False + + def _unexpected_selector(**_: object) -> _SelectedToolCall: + nonlocal model_called + model_called = True + return _SelectedToolCall(name="calendar_admin_create_event", arguments={}) + + result = run_live_model_turn( + user_intent=( + "Create an admin calendar event named Quarterly access review on " + "calendar ops-admin." + ), + authoritative_state=engine.state, + compiler_input="prohibit calendar_admin", + artifact_path=artifact_path, + model_tool_selector=_unexpected_selector, + ) + + assert result["decision_kind"] == "clarify" + assert result["executed"] is False + assert "currently in use" in (result["prompt_to_user"] or "") + assert model_called is False + assert _read_jsonl(artifact_path) == [] diff --git a/python/tests/test_mcp_calendar_admin_live_model_provider.py b/python/tests/test_mcp_calendar_admin_live_model_provider.py new file mode 100644 index 0000000..34696d8 --- /dev/null +++ b/python/tests/test_mcp_calendar_admin_live_model_provider.py @@ -0,0 +1,183 @@ +from types import SimpleNamespace + +import pytest + +from host_support.provider_mode import ProviderConfig +from python.examples._shared.litellm_request import build_litellm_provider_kwargs +from python.examples.tool_gating.mcp_calendar_admin import live_model as module + + +def test_build_litellm_provider_kwargs_prefixes_bare_ollama_model() -> None: + config = ProviderConfig( + mode="ollama", + source="PROVIDER", + base_url="http://localhost:11434", + model="qwen2.5:1.5b-instruct", + api_key=None, + ) + + assert build_litellm_provider_kwargs(config)["model"] == ( + "ollama/qwen2.5:1.5b-instruct" + ) + + +def test_build_litellm_provider_kwargs_preserves_prefixed_ollama_model() -> None: + config = ProviderConfig( + mode="ollama", + source="PROVIDER", + base_url="http://localhost:11434", + model="ollama/qwen2.5:1.5b-instruct", + api_key=None, + ) + + assert build_litellm_provider_kwargs(config)["model"] == ( + "ollama/qwen2.5:1.5b-instruct" + ) + + +def test_call_live_model_uses_normalized_ollama_model( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[dict[str, object]] = [] + + def fake_completion(**kwargs: object) -> dict[str, object]: + calls.append(dict(kwargs)) + return { + "choices": [ + { + "message": { + "tool_calls": [ + { + "function": { + "name": "calendar_admin_create_event", + "arguments": ( + '{"calendar_id":"ops-admin",' + '"event_title":"Quarterly access review"}' + ), + } + } + ] + } + } + ] + } + + monkeypatch.setattr( + module, + "import_module", + lambda name: SimpleNamespace(completion=fake_completion), + ) + monkeypatch.setattr( + module, + "resolve_provider_config", + lambda default_model: SimpleNamespace( + mode="ollama", + model="qwen2.5:1.5b-instruct", + base_url="http://127.0.0.1:11434", + api_key=None, + ), + ) + monkeypatch.setattr(module, "print_startup_config", lambda config: None) + monkeypatch.setenv("PROVIDER", "ollama") + + selected = module._call_live_model( + user_intent="Create the admin event.", + tools=[ + { + "type": "function", + "function": { + "name": "calendar_admin_create_event", + "description": "Create an event.", + "parameters": {"type": "object"}, + }, + } + ], + ) + + assert selected.name == "calendar_admin_create_event" + assert calls[0]["model"] == "ollama/qwen2.5:1.5b-instruct" + assert calls[0]["temperature"] == 0 + assert calls[0]["drop_params"] is True + + +def test_build_litellm_provider_kwargs_keeps_temperature_for_gpt5() -> None: + config = ProviderConfig( + mode="openai", + source="PROVIDER", + base_url="https://api.openai.com/v1", + model="gpt-5-mini", + api_key="test-key", + ) + + kwargs = build_litellm_provider_kwargs(config) + + assert kwargs["temperature"] == 0 + assert kwargs["drop_params"] is True + + +def test_build_litellm_provider_kwargs_keeps_temperature_for_gpt4o_mini() -> None: + config = ProviderConfig( + mode="openai", + source="PROVIDER", + base_url="https://api.openai.com/v1", + model="gpt-4o-mini", + api_key="test-key", + ) + + kwargs = build_litellm_provider_kwargs(config) + + assert kwargs["temperature"] == 0 + assert kwargs["drop_params"] is True + + +def test_build_litellm_provider_kwargs_keeps_temperature_for_ollama() -> None: + config = ProviderConfig( + mode="ollama", + source="PROVIDER", + base_url="http://localhost:11434", + model="qwen2.5:1.5b-instruct", + api_key=None, + ) + + kwargs = build_litellm_provider_kwargs(config) + + assert kwargs["temperature"] == 0 + assert kwargs["drop_params"] is True + + +def test_call_live_model_passes_temperature_and_drop_params_for_gpt5( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[dict[str, object]] = [] + + def fake_completion(**kwargs: object) -> dict[str, object]: + calls.append(dict(kwargs)) + return {"choices": [{"message": {"tool_calls": []}}]} + + monkeypatch.setattr( + module, + "import_module", + lambda name: SimpleNamespace(completion=fake_completion), + ) + monkeypatch.setattr( + module, + "resolve_provider_config", + lambda default_model: SimpleNamespace( + mode="openai", + model="gpt-5-mini", + base_url="https://api.openai.com/v1", + api_key="test-key", + ), + ) + monkeypatch.setattr(module, "print_startup_config", lambda config: None) + monkeypatch.delenv("PROVIDER", raising=False) + + selected = module._call_live_model( + user_intent="Create the admin event.", + tools=[], + ) + + assert selected.name is None + assert calls[0]["model"] == "gpt-5-mini" + assert calls[0]["temperature"] == 0 + assert calls[0]["drop_params"] is True