Build a client in Python

A working Zorlya MCP client fits in under 80 lines of Python using only the standard library. This guide builds one that calls tools, follows cursors and keeps the rotated refresh token. It runs on Python 3.9 or newer with nothing to install.

You need three values from Sign in with OAuth by hand: your client_id, an access token and a refresh token. Export them:

export ZORLYA_CLIENT_ID=…
export ZORLYA_ACCESS_TOKEN=…
export ZORLYA_REFRESH_TOKEN=…

The client

Save this as client.py:

"""A minimal Zorlya MCP client. Python 3.9 or newer, standard library only."""
import json
import os
import urllib.error
import urllib.parse
import urllib.request

# Two hosts: the authorization server (token endpoint) and the MCP server.
ISSUER = os.environ.get("ZORLYA_API", "https://api.zorlya.com")
MCP = os.environ.get("ZORLYA_MCP", "https://mcp.zorlya.com")
CLIENT_ID = os.environ["ZORLYA_CLIENT_ID"]
access_token = os.environ["ZORLYA_ACCESS_TOKEN"]
refresh_token = os.environ["ZORLYA_REFRESH_TOKEN"]
next_id = 1


def post(url, body, headers):
    request = urllib.request.Request(url, data=body, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request) as response:
            return response.status, json.load(response)
    except urllib.error.HTTPError as error:
        return error.code, json.load(error)


def refresh():
    global access_token, refresh_token
    form = urllib.parse.urlencode(
        {"grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": CLIENT_ID}
    ).encode()
    status, body = post(ISSUER + "/oauth/token", form, {"content-type": "application/x-www-form-urlencoded"})
    if status != 200:
        raise RuntimeError(f"refresh failed: {body['error']}: {body['error_description']}")
    # The refresh token rotates: keep the new one, the old one no longer works.
    access_token, refresh_token = body["access_token"], body["refresh_token"]


def rpc(method, params=None, retried=False):
    global next_id
    payload = {"jsonrpc": "2.0", "id": next_id, "method": method}
    if params is not None:
        payload["params"] = params
    next_id += 1
    status, message = post(
        MCP,
        json.dumps(payload).encode(),
        {"content-type": "application/json", "authorization": f"Bearer {access_token}"},
    )
    if status == 401 and not retried:
        refresh()
        return rpc(method, params, retried=True)
    if "error" in message:
        raise RuntimeError(f"{method}: {message['error']['code']} {message['error']['message']}")
    return message["result"]


def call_tool(name, **arguments):
    """A tool failure comes back as a result with isError, not as a JSON-RPC error."""
    result = rpc("tools/call", {"name": name, "arguments": arguments})
    text = result["content"][0]["text"]
    if result.get("isError"):
        raise RuntimeError(text)
    return json.loads(text)


def all_pages(name, key, **arguments):
    """Follows next_cursor until has_more is false."""
    items, cursor = [], None
    while True:
        page = call_tool(name, **arguments, **({"cursor": cursor} if cursor else {}))
        items += page[key]
        if not page["has_more"]:
            return items
        cursor = page["next_cursor"]


rpc("initialize", {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "my-client", "version": "0.1.0"}})
top = all_pages("top_matches", "matches", limit=20)
print(f"{len(top)} matches; best: {top[0]['title']} at {top[0]['company']} ({top[0]['score']})")

Run it with python3 client.py. It prints the number of matches and the best one.

How it works

• rpc sends one JSON-RPC message with the bearer token. On a 401 it refreshes once and retries. A second 401 raises, which means the person has to sign in again.
• refresh trades the refresh token for a new pair. The old one stops working as soon as this succeeds, so the code keeps the new one. If you store tokens, write the new pair before you use it.
• call_tool parses the text the tool returned and raises when isError is set, because a failed tool is a normal result and not a JSON-RPC error.
• all_pages follows next_cursor while has_more is true. See Page results and handle errors.

Make it your own

• Set ZORLYA_API and ZORLYA_MCP to test against a local copy of the server. The MCP address is the root of mcp.zorlya.com, with no path.
• Use requests or httpx if you prefer, and keep the headers and the body the same.
• Call rpc("tools/list") and read each inputSchema to validate arguments before you send them.
• Stay under 120 reads, 30 writes and 5 deletes a minute for the account.

What we checked

The first version of this file was run on Python 3.14 against a local copy of the server on 2026-10-17. On 2026-10-18 it was changed to use the two hosts, and the new version was checked for syntax only. It was not run against the production server.