#!/usr/bin/env python3 """Read-only TKP MCP example. Python 3.10+, no third-party dependencies.""" import argparse import json import os import sys import urllib.error import urllib.request class McpError(RuntimeError): pass class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # Never forward the bearer token to a redirected destination. return None class Client: VERSION = "2025-11-25" def __init__(self, url, token, timeout=60): if not url.startswith(("https://", "http://")): raise ValueError("Use an absolute HTTP(S) MCP URL") self.url = url self.token = token self.timeout = timeout self.session = None self.version = None self.sequence = 0 self.opener = urllib.request.build_opener(NoRedirect) def request(self, method, params=None, *, notify=False): self.sequence += 1 payload = {"jsonrpc": "2.0", "method": method} if params is not None: payload["params"] = params if not notify: payload["id"] = self.sequence headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "Accept": "application/json, text/event-stream", } if self.version: headers["MCP-Protocol-Version"] = self.version if self.session: headers["MCP-Session-Id"] = self.session request = urllib.request.Request( self.url, data=json.dumps(payload).encode(), headers=headers, method="POST" ) try: with self.opener.open(request, timeout=self.timeout) as response: self.session = response.headers.get("MCP-Session-Id", self.session) if notify: response.read() return None content_type = response.headers.get_content_type() if content_type == "text/event-stream": message = self._read_sse(response, payload["id"]) elif content_type == "application/json": message = json.load(response) else: raise McpError(f"Unexpected response type: {content_type}") except urllib.error.HTTPError as error: # Do not print bodies or signed URLs, which may contain secrets. raise McpError(f"HTTP {error.code}; see the MCP troubleshooting guide") from error except (urllib.error.URLError, TimeoutError) as error: raise McpError("MCP connection failed or timed out") from error if message.get("id") != payload["id"]: raise McpError("Response ID does not match request") if "error" in message: raise McpError(f"JSON-RPC error: {message['error']}") if "result" not in message: raise McpError("Response has no result") return message["result"] @staticmethod def _read_sse(response, request_id): data = [] for raw_line in response: line = raw_line.decode("utf-8").rstrip("\r\n") if not line: if data: message = json.loads("\n".join(data)) if message.get("id") == request_id: return message if "id" in message and "method" in message: raise McpError("This example does not handle server-initiated requests") data = [] elif line.startswith("data:"): data.append(line[5:].removeprefix(" ")) raise McpError("Stream ended without a matching response") def initialize(self): result = self.request("initialize", { "protocolVersion": self.VERSION, "capabilities": {}, "clientInfo": {"name": "tkp-read-example", "version": "1.0.0"}, }) if result.get("protocolVersion") != self.VERSION: raise McpError("Server did not negotiate the example's supported version") self.version = self.VERSION self.request("notifications/initialized", notify=True) return result def tools(self): tools = [] cursor = None seen = set() while True: result = self.request("tools/list", {"cursor": cursor} if cursor else {}) tools.extend(result["tools"]) cursor = result.get("nextCursor") if not cursor: return tools if cursor in seen: raise McpError("Server repeated a tools/list cursor") seen.add(cursor) def call(self, name, arguments): result = self.request("tools/call", {"name": name, "arguments": arguments}) if result.get("isError"): raise McpError(f"Tool {name} failed: {result.get('content', [])}") return result def structured(result): """TKP data tools return structuredContent; media tools also carry content blocks.""" if "structuredContent" not in result: raise McpError("Expected structuredContent from this data tool") return result["structuredContent"] def inspect_first(client, search): """Read one text asset from the first candidate; do not infer relevance.""" if not search["results"]: return {"notice": "No candidate to inspect"} resource_id = search["results"][0]["resource"]["id"] result = {"resource": structured(client.call("inspect_resource", {"resource_id": resource_id}))} assets = structured(client.call("list_resource_assets", {"resource_id": resource_id, "type": "text", "limit": 1})) if not assets["items"]: result["notice"] = "No text asset; use the media workflow for this resource" return result result["evidence"] = client.call("retrieve_content", {"target": {"kind": "asset", "asset_id": assets["items"][0]["id"]}}) reference = structured(result["evidence"])["source_reference"] result["citations"] = structured(client.call("resolve_citations", {"references": [reference]})) return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--query", help="Optional search; default only lists libraries") parser.add_argument("--library", help="Restrict search to one readable lib_ ID") parser.add_argument("--inspect-first", action="store_true", help="With --query, read one text asset of the first candidate and resolve its citation") args = parser.parse_args() if args.inspect_first and not args.query: parser.error("--inspect-first requires --query") url = os.environ.get("TKP_MCP_URL", "").strip() token = os.environ.get("TKP_TOKEN", "").strip() if not url or not token: parser.error("Set TKP_MCP_URL (including /mcp) and TKP_TOKEN") try: client = Client(url, token) server = client.initialize() tools = client.tools() names = {tool["name"] for tool in tools} if "list_libraries" not in names: raise McpError("list_libraries unavailable; grant read access to a library") output = {"server": server["serverInfo"], "tools": sorted(names), "libraries": []} cursor = None seen_cursors = set() while True: arguments = {"cursor": cursor} if cursor else {} page = structured(client.call("list_libraries", arguments)) output["libraries"].extend(page["items"]) cursor = page.get("next_cursor") if not cursor: break if cursor in seen_cursors: raise McpError("Server repeated a list_libraries cursor") seen_cursors.add(cursor) if args.query: arguments = {"query": args.query, "output_format": "compact"} if args.library: arguments["library_ids"] = [args.library] output["search"] = structured(client.call("search_resources", arguments)) if args.inspect_first: output["inspection"] = inspect_first(client, output["search"]) print(json.dumps(output, indent=2, ensure_ascii=False)) except (McpError, ValueError) as error: print(str(error), file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())