Spaces:
Paused
Paused
File size: 2,430 Bytes
8c1b9fe 656439d 8c1b9fe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | """End-to-end MCP protocol test: drive the auralynq-mcp server over stdio.
Unlike test_mcp.py (which calls the transport-agnostic tool functions directly),
this spins up the real server process and talks to it through the MCP stdio
transport — proving an external MCP client (Claude Desktop, IDEs, …) can connect,
list tools, and call them. Skipped automatically if the optional `mcp` SDK is
not installed, so it never breaks the $0 offline suite.
"""
from __future__ import annotations
import json
import os
import sys
import pytest
pytest.importorskip("mcp", reason="MCP SDK (optional 'mcp' extra) not installed")
pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
async def _run(corpus_dir, question):
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
env = dict(os.environ)
env.update(
{
"AURALYNQ_DATA_DIR": str(corpus_dir.parent),
"AURALYNQ_VECTOR__BACKEND": "memory",
"AURALYNQ_EMBEDDING__PROVIDER": "hash",
"AURALYNQ_LLM__PROVIDER": "extractive",
"AURALYNQ_VOICE__ASR_PROVIDER": "null",
"AURALYNQ_VOICE__TTS_PROVIDER": "null",
"AURALYNQ_LOG_LEVEL": "CRITICAL",
"AURALYNQ_DOTENV_DISABLED": "1",
}
)
params = StdioServerParameters(
command=sys.executable, args=["-m", "auralynq.mcp_server.server"], env=env
)
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
listed = await session.list_tools()
names = {t.name for t in listed.tools}
# index the corpus through the protocol, then query it
await session.call_tool("ingest_documents", {"path": str(corpus_dir)})
res = await session.call_tool("talk_to_data", {"question": question})
payload = json.loads(res.content[0].text)
return names, payload
async def test_mcp_stdio_lists_and_calls_tools(corpus_dir):
names, payload = await _run(corpus_dir, "What is the capital of France?")
assert {
"ingest_documents",
"search",
"graph_path_query",
"transcribe",
"talk_to_data",
"run_eval",
"get_trace",
"remember",
"recall",
} <= names
assert payload.get("answer")
assert "route" in payload
|