Guida a Vendooly

# Vendooly from a Python agent — MCP client with OAuth (pip install mcp)
import asyncio, threading, webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs

from pydantic import AnyUrl
from mcp import ClientSession
from mcp.client.auth import OAuthClientProvider, TokenStorage
from mcp.client.streamable_http import streamablehttp_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken

MCP_URL = "https://mcp.vendooly.com/mcp"
CALLBACK_PORT = 3000


class InMemoryTokenStorage(TokenStorage):
    # For production, persist tokens to disk/secret store instead.
    def __init__(self):
        self.tokens = None
        self.client_info = None

    async def get_tokens(self) -> OAuthToken | None:
        return self.tokens

    async def set_tokens(self, tokens: OAuthToken) -> None:
        self.tokens = tokens

    async def get_client_info(self) -> OAuthClientInformationFull | None:
        return self.client_info

    async def set_client_info(self, info: OAuthClientInformationFull) -> None:
        self.client_info = info


def wait_for_callback() -> str:
    """Tiny local web server: receives the consent redirect, returns the URL."""
    result = {}
    done = threading.Event()

    class H(BaseHTTPRequestHandler):
        def do_GET(self):
            result["url"] = self.path
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.end_headers()
            self.wfile.write(b"<h2>Authorized. You can close this tab.</h2>")
            done.set()

        def log_message(self, *a):
            pass

    srv = HTTPServer(("localhost", CALLBACK_PORT), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    done.wait(timeout=300)
    srv.shutdown()
    return result.get("url", "")


async def handle_redirect(auth_url: str) -> None:
    print(f"Opening browser for the Vendooly consent:\n{auth_url}")
    webbrowser.open(auth_url)


async def handle_callback() -> tuple[str, str | None]:
    path = wait_for_callback()
    q = parse_qs(urlparse(path).query)
    return q["code"][0], (q.get("state") or [None])[0]


oauth = OAuthClientProvider(
    server_url=MCP_URL,
    client_metadata=OAuthClientMetadata(
        client_name="My Python Agent",
        redirect_uris=[AnyUrl(f"http://localhost:{CALLBACK_PORT}/callback")],
        grant_types=["authorization_code", "refresh_token"],
        response_types=["code"],
    ),
    storage=InMemoryTokenStorage(),
    redirect_handler=handle_redirect,
    callback_handler=handle_callback,
)


async def main():
    async with streamablehttp_client(MCP_URL, auth=oauth) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("Available tools:", [t.name for t in tools.tools])
            res = await session.call_tool("list_my_products", {})
            print(res.content[0].text[:500])


if __name__ == "__main__":
    asyncio.run(main())