auth_required

Python · package toolnexus · SPEC §10 · python/src/toolnexus/types.py

def auth_required(
    url: str,
    prompt: str = "Authorization required to continue",
) -> ToolResult

Sugar over pending for the single most common suspension shape: pending(kind="authorization", prompt=prompt, url=url). By SPEC §10 convention, kind="authorization" follows OAuth2/OIDC authorization-code semantics — url is the authorize endpoint; the host’s wait_for performs the redirect → consent → callback out-of-band and may carry a resulting token in answer.data. toolnexus itself stays OIDC-agnostic: no OIDC library, no token logic in core — that all lives at the edge, inside your wait_for.

When to use it

A tool needs the caller to be logged in / to have granted a specific permission before it can do its job — a first API call that comes back 401, a scope your integration doesn’t have yet.
You want the “login” case to read as intent (auth_required(url)) rather than a generic pending(kind="authorization", ...) call every tool author has to spell out by hand.

Why this and not the alternative

Examples

1. The smallest useful call — the default prompt

from toolnexus import auth_required, pending_of

res = auth_required("https://example.com/oauth/authorize?client=toolnexus")

assert res.is_error is True
req = pending_of(res)
assert req.kind == "authorization"
assert req.url == "https://example.com/oauth/authorize?client=toolnexus"
assert req.prompt == "Authorization required to continue"  # the default
assert res.output == "Authorization required to continue\nhttps://example.com/oauth/authorize?client=toolnexus"

print("ok:", req.kind, "|", req.url)

2. The realistic case — a custom prompt naming what’s blocked

from toolnexus import auth_required, pending_of

res = auth_required(
    "https://example.com/oauth/authorize?scope=billing.read",
    prompt="Connect your billing account to check invoice status",
)

req = pending_of(res)
assert req.prompt == "Connect your billing account to check invoice status"
assert req.url == "https://example.com/oauth/authorize?scope=billing.read"
assert req.id  # still generated, exactly as pending() does

print("ok:", req.prompt)

3. The full surface — a tool that suspends for auth, then succeeds post-`wait_for`

The retry after wait_for resolves is for the same call: an authorization suspension usually doesn’t need ctx.answer.data at all — the session is valid now, so the tool just proceeds as if it always had access.

import asyncio
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from toolnexus import Answer, ToolResult, auth_required, create_client, create_toolkit, define_tool


class StubServer:
    def __init__(self, handler):
        outer = self

        class H(BaseHTTPRequestHandler):
            def log_message(self, *a):
                pass

            def do_POST(self):  # noqa: N802
                length = int(self.headers.get("Content-Length", 0))
                body = json.loads(self.rfile.read(length) or b"{}")
                outer._send(self, handler(body))

        self._server = ThreadingHTTPServer(("127.0.0.1", 0), H)
        self.port = self._server.server_address[1]
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)

    @staticmethod
    def _send(req, payload):
        body = json.dumps(payload).encode("utf-8")
        req.send_response(200)
        req.send_header("Content-Type", "application/json")
        req.send_header("Content-Length", str(len(body)))
        req.end_headers()
        req.wfile.write(body)

    @property
    def base_url(self) -> str:
        return f"http://127.0.0.1:{self.port}/v1"

    def __enter__(self):
        self._thread.start()
        return self

    def __exit__(self, *exc):
        self._server.shutdown()
        self._server.server_close()


# A minimal session model: "logged in" flips once wait_for has run.
SESSION = {"logged_in": False}


async def check_billing(args=None, ctx=None):
    if not SESSION["logged_in"]:
        return auth_required("https://example.com/oauth/authorize?scope=billing.read")
    return ToolResult(output="invoice #4821: paid", is_error=False)


def tool_call_reply(body):
    if not any(m.get("role") == "tool" for m in body["messages"]):
        return {
            "choices": [{
                "message": {
                    "role": "assistant", "content": None,
                    "tool_calls": [{"id": "c1", "type": "function",
                                     "function": {"name": "check_billing", "arguments": "{}"}}],
                },
                "finish_reason": "tool_calls",
            }],
            "usage": {"prompt_tokens": 6, "completion_tokens": 2, "total_tokens": 8},
        }
    return {
        "choices": [{"message": {"role": "assistant", "content": "your invoice is paid"}}],
        "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4},
    }


async def main():
    tk = await create_toolkit()
    tk.register(define_tool(check_billing, name="check_billing", description="Check invoice status."))
    try:

        async def wait_for(request):
            assert request.kind == "authorization"
            SESSION["logged_in"] = True  # the out-of-band redirect → consent → callback happened
            return Answer(id=request.id, ok=True)

        with StubServer(tool_call_reply) as srv:
            client = create_client(
                base_url=srv.base_url, style="openai", model="test-model", api_key="test-key",
                wait_for=wait_for,
            )
            result = await client.run("what's my invoice status?", tk)

        assert result.status == "done"
        assert result.tool_calls[0]["output"] == "invoice #4821: paid"
        assert SESSION["logged_in"] is True

        print("ok:", result.text, "|", result.tool_calls[0]["output"])
    finally:
        await tk.close()


asyncio.run(main())

Fields (the `Request` this builds)

Field	Value
`kind`	Always `"authorization"`.
`url`	The value you passed — the authorize endpoint.
`prompt`	Your `prompt`, or the default `"Authorization required to continue"`.
`id`	Generated.