Appearance
Build an MCP integration
Use this example to connect your application to TKP, discover its tools and read some evidence. It uses the Model Context Protocol without a chat client or language model. If you prefer ordinary HTTP endpoints, start with the REST API.
1. Prepare access
You need Python 3.10+, a running TKP instance and a Personal Access Token. Enable the MCP interface on the token and grant read access to a library containing a document whose processing has finished.
Use the TKP application address, for example https://tkp.example.com, rather than the address of this documentation site. The MCP endpoint is that address followed by /mcp.
2. List your libraries
Download client.py and open a terminal in the directory where you saved it. The example uses only Python's standard library, so no package installation is needed.
Run these commands in Bash or Zsh, substituting your MCP address:
bash
export TKP_MCP_URL='https://tkp.example.com/mcp'
read -rs TKP_TOKEN
export TKP_TOKEN
python3 client.pyAfter read -rs TKP_TOKEN, paste your token and press Enter. The terminal hides what you paste. Use HTTP rather than HTTPS only for local development.
The output is JSON containing the server name, available tool names and a libraries array. If the example reports that list_libraries is unavailable, check that the token grants read access to a library. Listing libraries does not call a model. Search in the next step uses the TKP instance's configured providers; your client needs only the TKP token.
3. Search and read a source
Choose a question relevant to your indexed documents and run:
bash
python3 client.py \
--query 'What does the project report say about energy consumption?' --inspect-firstThe example searches your readable libraries and inspects the first result. It then reads one text asset from that resource and resolves its source reference into a citation URL. An asset is a piece of a resource, such as extracted text or an image. Here, a resource is an uploaded document or media file.
The search output contains the matching candidates. The inspection output contains the selected resource, retrieved evidence and citation. The first result and first text asset may not answer your question: your application must check their relevance before making a claim. If the resource has no text asset, the example reports that instead of trying to interpret its media. See the media workflow for that next step.
To restrict the search, add --library lib_..., replacing lib_... with an actual library ID from step 2. Without that argument, all readable libraries are searched.
4. Use tools in your application
An MCP client library handles the connection and exchanges JSON messages with TKP. It first discovers available tools with tools/list. A response can span multiple pages; request the next page using its nextCursor until none remains. Each tool describes its accepted arguments in inputSchema.
The client invokes a selected tool using tools/call. For example, this message asks TKP to search without returning asset contents yet:
json
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "search_resources",
"arguments": {"query": "energy consumption", "output_format": "compact"}
}
}A successful data tool returns its data in result.structuredContent. Media tools also return result.content: an array of text, image, audio or file-link blocks that your application must render or pass to a model. Keep both fields; reading only the metadata would discard the retrieved content.
Check result.isError before using a result. A tool can report failure even when the HTTP request itself succeeded. Tool discovery also depends on the credential: discover tools for each user's connection rather than reusing another user's tool list as permission evidence.
Use an MCP client library for a production integration, particularly when you need OAuth, reconnection or protocol upgrades. The example below shows the HTTP exchange directly so you can inspect it without installing that library.
Example source
Download client.py, or inspect the source below. In a repository checkout the same file is at services/docs/examples/mcp/client.py.
Complete source
python
#!/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())Protocol version used by this example
The example uses TKP's supported 2025-11-25 protocol version. It sends an initialize request, checks the negotiated version, then sends notifications/initialized before calling tools. Later requests carry the protocol version and any session header returned by the server.
It accepts responses encoded as JSON or Server-Sent Events (SSE). These are two HTTP response formats for the same MCP messages. Newer client libraries may negotiate a newer protocol version with different connection rules. See the MCP transport specification when implementing the transport yourself.
Next steps
- Workflows: inspect evidence, resolve citations, upload and poll jobs.
- Troubleshooting: diagnose missing tools and failed calls.
- Tool reference: look up arguments and response schemas.