diff --git a/README.md b/README.md index 7bcf0a5..8e6a474 100644 --- a/README.md +++ b/README.md @@ -337,9 +337,11 @@ Python 3.11+ can connect directly with the [official MCP Python SDK](https://git ```bash uv run examples/mcp_docs.py +uv run examples/mcp_docs.py --query "Robinhood mainnet RPC" +uv run examples/mcp_docs.py --query "Robinhood 主网 RPC" --lang zh ``` -The example pins `mcp==2.3.0` in its script metadata, lists the two documentation tools, then reads the English `quickstart` page. It prints JSON containing the tool names, page path, language and complete Markdown text, so an agent can consume the result. It uses the documentation-only endpoint: no BlockVectra key, wallet, model API or chain RPC is involved. Errors exit non-zero; the example has a 40-second overall deadline. Dependency installation occurs before that deadline. +The example pins `mcp==2.3.0` in its script metadata, lists the two documentation tools, then reads the English `quickstart` page by default. With `--query`, it searches by keyword and reads only the first matching document in the selected `--lang`; no matches cause an error instead of silently reading `quickstart`. Without `--query`, `--lang` selects the language of `quickstart`. It prints JSON containing the tool names, page path, language and complete Markdown text, so an agent can consume the result. It uses the documentation-only endpoint: no BlockVectra key, wallet, model API or chain RPC is involved. Errors exit non-zero; the example has a 40-second overall deadline. Dependency installation occurs before that deadline. The separate standard-library RPC example above still needs no third-party Python packages. diff --git a/README.zh.md b/README.zh.md index 21da4b3..70f085b 100644 --- a/README.zh.md +++ b/README.zh.md @@ -337,9 +337,11 @@ Python 3.11+ 可通过[官方 MCP Python SDK](https://github.com/modelcontextpro ```bash uv run examples/mcp_docs.py +uv run examples/mcp_docs.py --query "Robinhood 主网 RPC" +uv run examples/mcp_docs.py --query "Robinhood 主网 RPC" --lang zh ``` -示例在脚本元数据中固定 `mcp==2.3.0`，先列出两个文档工具，再读取英文 `quickstart` 页。输出 JSON 包含工具名、页面路径、语言和完整 Markdown，Agent 可直接消费。它只连接文档专用入口，不需要 BlockVectra key、钱包或模型 API，也不调用链上 RPC。出错时非零退出；示例运行总时限为 40 秒，依赖安装不包含在该时限内。 +示例在脚本元数据中固定 `mcp==2.3.0`，先列出两个文档工具，默认再读取英文 `quickstart` 页。传入 `--query` 时，先按关键词检索，再读取所选 `--lang` 的首条匹配文档；无匹配时报错，不会悄悄改读 `quickstart`。不传 `--query` 时，`--lang` 可选择 `quickstart` 的语言。输出 JSON 包含工具名、页面路径、语言和完整 Markdown，Agent 可直接消费。它只连接文档专用入口，不需要 BlockVectra key、钱包或模型 API，也不调用链上 RPC。出错时非零退出；示例运行总时限为 40 秒，依赖安装不包含在该时限内。 前面的独立标准库 RPC 示例仍无需第三方 Python 包。 diff --git a/examples/mcp_docs.py b/examples/mcp_docs.py index 18df29e..01888b0 100644 --- a/examples/mcp_docs.py +++ b/examples/mcp_docs.py @@ -4,9 +4,12 @@ # /// """Read BlockVectra documentation with the official Python MCP SDK.""" +import argparse import asyncio import json import os +import unicodedata +from urllib.parse import urlsplit import httpx2 from mcp import ClientSession @@ -15,7 +18,45 @@ ENDPOINT = "https://docs.blockvectra.com/mcp/docs" -async def read_quickstart(http_client: httpx2.AsyncClient) -> dict: +def tool_text(result, tool: str) -> str: + payload = result.model_dump(mode="json", by_alias=True) + if payload.get("isError", False): + raise RuntimeError(f"{tool} returned a tool error.") + texts = [item["text"] for item in payload.get("content", []) if item.get("type") == "text"] + if not texts or not any(text.strip() for text in texts): + raise RuntimeError(f"{tool} returned no documentation text.") + return "\n".join(texts) + + +def first_path(text: str) -> str: + try: + payload = json.loads(text) + except json.JSONDecodeError as exc: + raise RuntimeError("search_docs returned invalid JSON.") from exc + if not isinstance(payload, dict) or not isinstance(payload.get("results"), list): + raise RuntimeError("search_docs returned invalid results.") + results = payload["results"] + if not results: + raise RuntimeError("search_docs returned no matching documents.") + path = results[0].get("path") if isinstance(results[0], dict) else None + if ( + not isinstance(path, str) + or not path + or path.startswith("/") + or "\\" in path + or "?" in path + or "#" in path + or ".." in path.split("/") + or any(char.isspace() or unicodedata.category(char) == "Cc" for char in path) + ): + raise RuntimeError("search_docs returned an invalid document path.") + parsed = urlsplit(path) + if parsed.scheme or parsed.netloc: + raise RuntimeError("search_docs returned an invalid document path.") + return path + + +async def read_document(http_client: httpx2.AsyncClient, query: str | None, lang: str) -> dict: async with streamable_http_client(ENDPOINT, http_client=http_client) as (read, write): async with ClientSession(read, write, read_timeout_seconds=15) as session: await session.initialize() @@ -23,24 +64,38 @@ async def read_quickstart(http_client: httpx2.AsyncClient) -> dict: names = sorted(tool.name for tool in available.tools) if names != ["read_doc", "search_docs"]: raise RuntimeError("The endpoint did not expose the two documentation tools.") - result = await session.call_tool("read_doc", {"path": "quickstart", "lang": "en"}) - payload = result.model_dump(mode="json", by_alias=True) - if payload.get("isError", False): - raise RuntimeError("read_doc returned a tool error.") - texts = [item["text"] for item in payload.get("content", []) if item.get("type") == "text"] - if not texts or not any(text.strip() for text in texts): - raise RuntimeError("read_doc returned no documentation text.") - return {"tools": names, "path": "quickstart", "lang": "en", "text": "\n".join(texts)} + path = "quickstart" + if query is not None: + search = await session.call_tool( + "search_docs", {"query": query, "lang": lang, "limit": 1} + ) + path = first_path(tool_text(search, "search_docs")) + result = await session.call_tool("read_doc", {"path": path, "lang": lang}) + output = {"tools": names, "path": path, "lang": lang, "text": tool_text(result, "read_doc")} + if query is not None: + output["query"] = query + return output + + +def nonempty(value: str) -> str: + value = value.strip() + if not value: + raise argparse.ArgumentTypeError("must not be empty or whitespace") + return value async def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--query", type=nonempty, metavar="TOPIC", help="search and read the first match") + parser.add_argument("--lang", type=nonempty, default="en", metavar="LANG", help="document language (default: en)") + args = parser.parse_args() user_agent = "BlockVectraDocsPython/1" if os.environ.get("BLOCKVECTRA_INTERNAL_QA") == "1": user_agent += " BlockVectraQA/1" async with httpx2.AsyncClient( timeout=15, follow_redirects=False, headers={"User-Agent": user_agent} ) as http_client: - result = await read_quickstart(http_client) + result = await read_document(http_client, args.query, args.lang) print(json.dumps(result, ensure_ascii=False, indent=2)) diff --git a/examples/test_mcp_docs.py b/examples/test_mcp_docs.py new file mode 100644 index 0000000..5f42b20 --- /dev/null +++ b/examples/test_mcp_docs.py @@ -0,0 +1,131 @@ +"""Offline behavior tests for the Python documentation example.""" + +import asyncio +from contextlib import ExitStack, redirect_stdout, redirect_stderr +import io +import json +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +# Block network access before importing the SDK or the example. +_network_guard = ExitStack() +for target in ("socket.create_connection", "socket.socket.connect", "socket.socket.connect_ex", "socket.getaddrinfo"): + _network_guard.enter_context(patch(target, side_effect=AssertionError("Network disabled in tests"))) + +import mcp_docs +from mcp.types import CallToolResult, TextContent + + +def tearDownModule(): + _network_guard.close() + + +def result(text=None, error=False): + return CallToolResult( + content=[] if text is None else [TextContent(type="text", text=text)], isError=error + ) + + +class McpDocsTests(unittest.TestCase): + def setUp(self): + self.stack = ExitStack() + self.addCleanup(self.stack.close) + self.http = self.stack.enter_context(patch.object(mcp_docs.httpx2, "AsyncClient")) + self.transport = self.stack.enter_context(patch.object(mcp_docs, "streamable_http_client")) + self.transport.return_value.__aenter__ = AsyncMock(return_value=(object(), object())) + self.transport.return_value.__aexit__ = AsyncMock(return_value=False) + self.session_factory = self.stack.enter_context(patch.object(mcp_docs, "ClientSession")) + self.session = MagicMock() + self.session.initialize = AsyncMock() + self.session.list_tools = AsyncMock(return_value=MagicMock(tools=[ + MagicMock(name="read"), MagicMock(name="search") + ])) + self.session.list_tools.return_value.tools[0].name = "read_doc" + self.session.list_tools.return_value.tools[1].name = "search_docs" + self.session.call_tool = AsyncMock(return_value=result("# Quickstart\n[source](https://docs.blockvectra.com)")) + self.session_factory.return_value.__aenter__ = AsyncMock(return_value=self.session) + self.session_factory.return_value.__aexit__ = AsyncMock(return_value=False) + + def run_cli(self, *args): + output = io.StringIO() + with patch("sys.argv", ["mcp_docs.py", *args]), redirect_stdout(output): + asyncio.run(mcp_docs.main()) + return json.loads(output.getvalue()) + + def test_default_reads_quickstart_without_search(self): + output = self.run_cli() + self.session.call_tool.assert_awaited_once_with("read_doc", {"path": "quickstart", "lang": "en"}) + self.assertEqual(output, { + "tools": ["read_doc", "search_docs"], "path": "quickstart", "lang": "en", + "text": "# Quickstart\n[source](https://docs.blockvectra.com)" + }) + + def test_language_without_query_is_forwarded(self): + output = self.run_cli("--lang", "fr") + self.session.call_tool.assert_awaited_once_with("read_doc", {"path": "quickstart", "lang": "fr"}) + self.assertEqual(output["lang"], "fr") + self.assertNotIn("query", output) + + def test_query_reads_only_first_match_in_same_language(self): + for path in ("guides/robinhood-chain", "reference/v1.2_slug+name", "指南/入门"): + with self.subTest(path=path): + self.session.call_tool.reset_mock() + self.session.call_tool.side_effect = [ + result(json.dumps({"results": [{"title": "First", "path": path, "summary": "RPC"}, {"path": "other"}]})), + result("# Document\n[source](https://docs.blockvectra.com/zh/guide)") + ] + output = self.run_cli("--query", " Robinhood 主网 RPC ", "--lang", "zh") + self.assertEqual(self.session.call_tool.await_args_list, [ + unittest.mock.call("search_docs", {"query": "Robinhood 主网 RPC", "lang": "zh", "limit": 1}), + unittest.mock.call("read_doc", {"path": path, "lang": "zh"}) + ]) + self.assertEqual(output["path"], path) + self.assertEqual(output["lang"], "zh") + self.assertEqual(output["query"], "Robinhood 主网 RPC") + self.assertEqual(output["text"], "# Document\n[source](https://docs.blockvectra.com/zh/guide)") + + def test_bad_search_results_never_read(self): + texts = [None, "", " \n", "not JSON", "[]", "null", "1", '"text"', "{}", + '{"results": null}', '{"results": {}}', '{"results": []}', + '{"results": [null]}', '{"results": ["path"]}', '{"results": [{}]}'] + paths = [None, 1, [], {}, "", " ", "/guides/doc", "//example.com/doc", + "https://example.com/doc", "https:doc", "../doc", "guides/../doc", + "guides/..", "guides\\doc", "doc?query", "doc#fragment", "doc\x00", + "doc\x7f", "doc\n", "doc name", "doc\u00a0name"] + bad_results = [result(text) for text in texts] + bad_results += [result(json.dumps({"results": [{"path": path}, {"path": "valid"}]})) for path in paths] + bad_results += [result('{"results": [{"path": "valid"}]}', error=True)] + for index, response in enumerate(bad_results): + with self.subTest(case=index): + self.session.call_tool.reset_mock() + self.session.call_tool.return_value = response + with self.assertRaises(RuntimeError): + self.run_cli("--query", "topic") + self.session.call_tool.assert_awaited_once_with("search_docs", {"query": "topic", "lang": "en", "limit": 1}) + + def test_read_errors_and_empty_text_fail(self): + for query_args in ((), ("--query", "topic")): + for response in (result("error", error=True), result(), result(""), result(" \n")): + with self.subTest(args=query_args, response=response): + self.session.call_tool.reset_mock() + self.session.call_tool.side_effect = ( + [result('{"results": [{"path": "guide"}]}'), response] if query_args else [response] + ) + with self.assertRaises(RuntimeError): + self.run_cli(*query_args) + self.assertEqual(self.session.call_tool.await_count, 2 if query_args else 1) + + def test_empty_cli_arguments_fail_before_network(self): + for option in ("--query", "--lang"): + for value in ("", " \t\n"): + with self.subTest(option=option, value=value), redirect_stderr(io.StringIO()): + with self.assertRaises(SystemExit) as raised: + self.run_cli(option, value) + self.assertEqual(raised.exception.code, 2) + self.http.assert_not_called() + self.transport.assert_not_called() + self.session_factory.assert_not_called() + + +if __name__ == "__main__": + unittest.main()