feat(osvx-mcp): expose eight read-only tools and verify discovery

This commit is contained in:
Chris Christiansen 2026-09-01 15:28:48 +00:00
parent aa7b40c8aa
commit 57fdf09a44
2 changed files with 130 additions and 0 deletions

View File

@ -369,6 +369,36 @@ async def push_file(p):
TOOLS = {
# Billing
"get_billing_summary": (get_billing_summary, "Hent billing-sammendrag for OPAX", {}),
"get_billing_forecast": (
get_billing_forecast,
"Get billing forecast",
{},
),
"get_billing_credits": (
get_billing_credits,
"Get available billing credits",
{},
),
"get_billing_anomalies": (
get_billing_anomalies,
"Get billing anomalies",
{},
),
"get_billing_history": (
get_billing_history,
"Get billing history",
{},
),
"get_billing_budget": (
get_billing_budget,
"Get the configured billing budget",
{},
),
"get_telemetry": (
get_billing_tokens_by_module,
"Get token and telemetry history by module",
{},
),
"set_billing_budget": (set_billing_budget, "Sett månedlig budsjett", {"type":"object","properties":{"amount":{"type":"number"}}}),
# Onboarding
"create_invite": (create_invite, "Inviter ny kunde til OPAX", {"type":"object","properties":{"company":{"type":"string"},"email":{"type":"string"},"tier":{"type":"string"}},"required":["email"]}),
@ -376,9 +406,22 @@ TOOLS = {
"send_webhook": (send_webhook, "Send webhook-varsling", {"type":"object","properties":{"url":{"type":"string"},"message":{"type":"string"}}}),
"send_email": (send_email, "Send e-post via ekstern tjeneste", {"type":"object","properties":{"to":{"type":"string"},"subject":{"type":"string"}},"required":["to","subject"]}),
"send_sms": (send_sms, "Send SMS", {"type":"object","properties":{"to":{"type":"string"},"message":{"type":"string"}},"required":["to","message"]}),
"get_notify_channels": (
get_notify_channels,
"List configured notification channels",
{},
),
# AI Agents
"run_jason": (run_jason, "Kjør Jason-agenten med en prompt", {"type":"object","properties":{"prompt":{"type":"string"},"mode":{"type":"string"}},"required":["prompt"]}),
"run_emma": (run_emma, "Emma Vauger (gemma3:27b) — primær lokal AI", {"type":"object","properties":{"prompt":{"type":"string"}}}),
# Local models — read-only discovery
"list_emma_models": (
list_emma_models,
"List locally available OSVx model names",
{},
),
# System & Ops
"get_health": (get_health, "Hent helsestatus for OPAX", {}),
"get_build_status": (get_build_status, "Hent siste build-status", {}),

View File

@ -0,0 +1,87 @@
from __future__ import annotations
import asyncio
import json
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
from typing import Any
import logging
SERVER_PATH = Path(__file__).with_name("server.py")
spec = spec_from_file_location("osvx_mcp_server_under_test", SERVER_PATH)
if spec is None or spec.loader is None:
raise RuntimeError(f"Cannot load server module from {SERVER_PATH}")
server = module_from_spec(spec)
# Temporarily suppress INFO logs from the server module during import
target_logger = logging.getLogger("osvx_mcp_server_under_test")
original_level = target_logger.level
target_logger.setLevel(logging.WARNING)
try:
spec.loader.exec_module(server)
finally:
target_logger.setLevel(original_level)
class FakeRequest:
def __init__(self, payload: dict[str, Any]) -> None:
self._payload = payload
async def json(self) -> dict[str, Any]:
return self._payload
async def rpc(method: str, request_id: str) -> dict[str, Any]:
response = await server.mcp_handler(
FakeRequest(
{
"jsonrpc": "2.0",
"method": method,
"id": request_id,
}
)
)
assert response.status_code == 200
payload = json.loads(response.body.decode("utf-8"))
assert payload["jsonrpc"] == "2.0"
assert payload["id"] == request_id
assert "result" in payload
return payload["result"]
async def verify() -> None:
initialize = await rpc("initialize", "initialize-test")
assert initialize["protocolVersion"] == "2024-11-05"
assert initialize["capabilities"]["tools"] == {}
assert initialize["serverInfo"]["name"] == "opax-mcp"
assert isinstance(initialize["serverInfo"]["version"], str)
assert initialize["serverInfo"]["version"]
discovery = await rpc("tools/list", "tools-list-test")
tools = discovery["tools"]
discovered_names = {tool["name"] for tool in tools}
assert len(server.TOOLS) == 32
assert len(tools) == 32
assert len(discovered_names) == 32
assert discovered_names == set(server.TOOLS)
for tool in tools:
assert isinstance(tool["name"], str) and tool["name"]
assert isinstance(tool["description"], str) and tool["description"]
assert isinstance(tool["inputSchema"], dict)
assert tool["inputSchema"].get("type") == "object"
print("OSVX_MCP_DISCOVERY_VERIFIED")
print("TOOLS_COUNT=32")
print("INITIALIZE_TOOLS_CAPABILITY=YES")
print("TOOLS_LIST_MATCHES_REGISTRY=YES")
if __name__ == "__main__":
asyncio.run(verify())