feat(examples): search documentation before reading with Python

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.
 

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 包。
 

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,32 +18,84 @@
 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()
             available = await session.list_tools()
             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))
 
 

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()