Token Broker Guide

Table of Contents

1. Prerequisites
2. Installation
3. Quick Start
4. Authentication Methods
5. Basic Usage Patterns
6. Working with Tools
7. Token Response Format
8. Making API Calls
9. Configuration Options
10. Properties and Methods
11. Error Handling
12. Best Practices
13. Complete Examples
14. MCP Client vs Token Client

Prerequisites

Before using the Token Client, ensure you have completed the following setup:

1. Register and Activate an Application

Applications act as the identity for your agent. Each application has credentials that allow it to access tools on behalf of users.

Option A: Register via Dashboard (Recommended)

1. Go to the Agentic Fabriq dashboard
2. Navigate to admin console (if you are an admin)
3. Click Register to receive an activation token and activate the app
4. Copy the App ID and App Secret from the activation screen

Option B: Register via CLI

See CLI Reference (afctl) for tags, parameters, etc.

After activation, you'll have:

• app_id: Your application identifier (e.g., org-abc123_my-api-agent)
• app_secret: Your application secret key (e.g., sk_live_xyz789...)

If using the CLI, credentials are saved to ~/.af/applications/{app_id}.json.

2. User Authentication

Users must authenticate before the Token Broker can access their tools. The authentication method depends on your deployment:

3. Tool Connections

Users must connect the tools they want to use. This is done via:

• Dashboard UI: Users connect tools in the Agentic Fabriq dashboard
• CLI: afctl tools add and afctl tools connect

The Token Client can only access tools that:

1. The user has connected
2. The application has scopes for

Installation

pip install agentic-fabriq-sdkcopy

Import the client:

from af_sdk import TokenClientcopy

Quick Start

import asyncio
import httpx
from af_sdk import TokenClient

async def main():
    async with TokenClient(
        method="cli",
        app_id="org-abc123_my-api-agent",
        app_secret="sk_live_xyz789..."
    ) as client:
        # List available tools (these can be passed to an LLM as function definitions)
        tools = await client.list_tools()
        print(f"Available tools: {client.tool_names}")
        
        # In a real agent, tool_name and tool_args come from the LLM's decision,
        # not hardcoded. The LLM decides which tool to call based on the user's request.
        tool_name = "google_gmail_list_messages"  # LLM chooses this
        tool_args = {"maxResults": 5}             # LLM provides these
        
        # Get OAuth token + API metadata from Token Broker
        # This does NOT execute the tool - it returns the info needed to call the API
        token = await client.get_token(tool_name, tool_args)
        
        # Make the API call yourself using the token and API info
        async with httpx.AsyncClient() as http:
            response = await http.get(
                token.full_url,      # Full URL with query params
                headers=token.headers # Includes Authorization: Bearer <token>
            )
            data = response.json()   # This is the actual API response
            print(f"Found {len(data.get('messages', []))} messages")

asyncio.run(main())copy

Authentication Methods

The Token Client supports three authentication methods, each suited for different deployment scenarios.

CLI Method (method="cli")

Use case: Local development, testing, CLI scripts, or any scenario where the user has already authenticated via afctl auth login.

How it works: The client reads stored credentials from ~/.af/credentials.json (created by afctl auth login) and exchanges them for an AF token.

Required parameters: app_id, app_secret

# User must first run: afctl auth login

async with TokenClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789..."
) as client:
    token = await client.get_token("google_gmail_list_messages", {"maxResults": 10})copy

Keycloak Method (method="keycloak")

Use case: Web applications or services that authenticate users directly through Keycloak (Agentic Fabriq's identity provider).

Required parameters: app_id, app_secret, keycloak_token

# Token obtained from your Keycloak OAuth flow
keycloak_token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."

async with TokenClient(
    method="keycloak",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789...",
    keycloak_token=keycloak_token
) as client:
    token = await client.get_token("slack_list_channels", {"limit": 50})copy

IdP Method (method="idp")

Use case: Enterprise organizations using external identity providers like Okta for SSO.

Required parameters: app_id, app_secret, external_token

Optional parameters: org_url (auto-detected from app if not provided, used to identify organization)

# Token obtained from Okta OAuth flow
okta_token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."

async with TokenClient(
    method="idp",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789...",
    external_token=okta_token,
    org_url="mycompany.com"  # Optional
) as client:
    token = await client.get_token("github_list_repos", {"per_page": 10})copy

Basic Usage Patterns

Async Context Manager (Recommended)

The async context manager automatically handles connection and cleanup:

from af_sdk import TokenClient

async with TokenClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789..."
) as client:
    # Client is connected and authenticated
    tools = await client.list_tools()
    token = await client.get_token("google_gmail_list_messages", {"maxResults": 10})
    # Connection automatically closed on exitcopy

Sync Context Manager

For synchronous code, use the sync context manager with _sync method variants:

from af_sdk import TokenClient

with TokenClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789..."
) as client:
    tools = client.list_tools_sync()
    token = client.get_token_sync("slack_list_channels", {"limit": 20})copy

Manual Connection Management

For fine-grained control over the connection lifecycle:

from af_sdk import TokenClient

client = TokenClient(
    method="cli",
    app_id="org-abc123_my-agent",
    app_secret="sk_live_xyz789..."
)

try:
    # Explicitly connect
    await client.connect()
    
    # Use the client
    tools = await client.list_tools()
    token = await client.get_token("google_drive_list_files", {"pageSize": 10})
finally:
    # Always disconnect
    await client.disconnect()copy

Working with Tools

Listing Available Tools

The list_tools() method fetches all tools available to the current user based on their connections and the application's scopes:

async with TokenClient(method="cli", app_id="...", app_secret="...") as client:
    tools = await client.list_tools()
    
    print(f"Found {len(tools)} tools")
    for tool in tools:
        print(f"\nTool: {tool['name']}")
        print(f"  Description: {tool['description']}")
        print(f"  Parameters: {tool['inputSchema']}")copy

Each tool definition includes:

• name: The tool identifier (e.g., google_gmail_list_messages)
• description: Human-readable description of what the tool does
• inputSchema: JSON Schema defining the tool's parameters

Accessing Cached Tools

After calling list_tools(), tools are cached for quick access:

async with TokenClient(method="cli", app_id="...", app_secret="...") as client:
    await client.list_tools()  # Populates cache
    
    # Quick access to tool names
    print(client.tool_names)
    # ['google_gmail_list_messages', 'google_drive_list_files', 'slack_list_channels', ...]
    
    # Check if a specific tool is available
    if client.has_tool("google_gmail_list_messages"):
        print("Gmail is available!")
    
    # Get full tool definition
    tool = client.get_tool("slack_list_channels")
    if tool:
        print(f"Slack tool schema: {tool['inputSchema']}")
    
    # Get all cached tools (no network call)
    all_tools = client.get_tools()copy

Getting Tokens for Tools

Use get_token() with the tool name and a dictionary of arguments:

token = await client.get_token("tool_name", {
    "param1": "value1",
    "param2": "value2"
})copy

The token response includes everything you need to make the API call yourself.

Token Response Format

When you call get_token(), you receive a TokenInfo object with all the information needed to make the API call:

token = await client.get_token("google_gmail_list_messages", {"maxResults": 10})

# Token fields
token.access_token       # str: OAuth access token
token.token_type         # str: "Bearer"
token.expires_in         # int: Seconds until expiration
token.expires_at         # str: ISO timestamp of expiration
token.provider           # str: Provider name (e.g., "google_gmail")
token.scopes             # List[str]: Granted OAuth scopes

# API call specification
token.method             # str: HTTP method ("GET", "POST", etc.)
token.base_url           # str: Provider base URL
token.endpoint           # str: API endpoint path
token.full_url           # str: Complete URL with query params
token.headers            # Dict[str, str]: Headers including Authorization
token.query_params       # Dict[str, Any]: Query parameters
token.body_template      # Dict | None: Body template for POST/PUT
token.body_instructions  # str | None: Instructions for building body

# Example
token.example_curl       # str: Ready-to-use curl commandcopy

Example Token Response

token = await client.get_token("google_gmail_list_messages", {"maxResults": 5, "q": "is:unread"})

print(token.full_url)
# https://gmail.googleapis.com/gmail/v1/users/me/messages?maxResults=5&q=is%3Aunread

print(token.headers)
# {'Authorization': 'Bearer ya29.a0AfH6...', 'Content-Type': 'application/json'}

print(token.example_curl)
# curl -X GET \
#   -H 'Authorization: Bearer ya29.a0AfH6...' \
#   -H 'Content-Type: application/json' \
#   'https://gmail.googleapis.com/gmail/v1/users/me/messages?maxResults=5&q=is%3Aunread'copy

Making API Calls

With httpx (Recommended)

import httpx
from af_sdk import TokenClient

async with TokenClient(method="cli", app_id="...", app_secret="...") as client:
    token = await client.get_token("google_gmail_list_messages", {"maxResults": 10})
    
    async with httpx.AsyncClient() as http:
        response = await http.get(token.full_url, headers=token.headers)
        data = response.json()copy

With requests (Sync)

import requests
from af_sdk import TokenClient

with TokenClient(method="cli", app_id="...", app_secret="...") as client:
    token = client.get_token_sync("slack_list_channels", {"limit": 20})
    
    response = requests.get(token.full_url, headers=token.headers)
    channels = response.json()["channels"]copy

POST Requests with Body

async with TokenClient(method="cli", app_id="...", app_secret="...") as client:
    token = await client.get_token("slack_send_message", {
        "channel": "C1234567890",
        "text": "Hello from Token Broker!"
    })
    
    # Build body from parameters
    body = {
        "channel": "C1234567890",
        "text": "Hello from Token Broker!"
    }
    
    async with httpx.AsyncClient() as http:
        response = await http.post(
            token.full_url,
            headers=token.headers,
            json=body
        )copy

Configuration Options

Properties and Methods

Properties

Connection Methods

Tool Methods

Error Handling

The SDK provides specific exception types for different error scenarios:

from af_sdk import TokenClient
from af_sdk.exceptions import (
    AuthenticationError,
    MCPConnectionError,
    MCPError
)

try:
    async with TokenClient(
        method="cli",
        app_id="org-abc123_my-agent",
        app_secret="sk_live_xyz789..."
    ) as client:
        token = await client.get_token("google_gmail_list_messages", {"maxResults": 10})
        
except AuthenticationError as e:
    # Authentication failed
    print(f"Authentication failed: {e}")
    print("Suggestion: Run 'afctl auth login' or check your app credentials")
    
except MCPConnectionError as e:
    # Connection failed
    print(f"Connection failed: {e}")
    print("Suggestion: Check network connectivity and broker URL")
    
except MCPError as e:
    # Token Broker returned an error
    print(f"Token Broker error: {e.message}")
    print(f"Error code: {e.code}")copy

Exception Reference

Best Practices

1. Always Use Context Managers

Context managers ensure proper cleanup even if errors occur:

# Good
async with TokenClient(...) as client:
    token = await client.get_token(...)

# Avoid (requires manual cleanup)
client = TokenClient(...)
await client.connect()
# If an error occurs here, disconnect() may not be called
token = await client.get_token(...)
await client.disconnect()copy

2. Check Tool Availability Before Requesting Tokens

async with TokenClient(...) as client:
    await client.list_tools()
    
    if not client.has_tool("google_gmail_list_messages"):
        print("Gmail not available - user may need to connect it")
        return
    
    token = await client.get_token("google_gmail_list_messages", {...})copy

3. Handle Token Expiration

Tokens have an expiration time. For long-running applications, check token.expires_in:

token = await client.get_token("google_gmail_list_messages", {})

if token.expires_in < 60:
    # Token expires in less than a minute, request a fresh one
    token = await client.get_token("google_gmail_list_messages", {})copy

4. Reuse HTTP Clients

async with TokenClient(...) as client:
    async with httpx.AsyncClient() as http:
        # Reuse the HTTP client for multiple requests
        token1 = await client.get_token("google_gmail_list_messages", {})
        response1 = await http.get(token1.full_url, headers=token1.headers)
        
        token2 = await client.get_token("google_drive_list_files", {})
        response2 = await http.get(token2.full_url, headers=token2.headers)copy

5. Use Appropriate Auth Method

Complete Examples

Gmail Reader

import asyncio
import httpx
from af_sdk import TokenClient
from af_sdk.exceptions import AuthenticationError, MCPError

async def list_unread_emails():
    """Fetch and display unread emails using the Token Broker."""
    
    try:
        async with TokenClient(
            method="cli",
            app_id="org-abc123_email-reader",
            app_secret="sk_live_xyz789..."
        ) as client:
            # Check what tools are available
            await client.list_tools()
            
            if not client.has_tool("google_gmail_list_messages"):
                print("Error: Gmail not connected")
                return
            
            # Get token for listing messages
            token = await client.get_token(
                "google_gmail_list_messages",
                {"maxResults": 10, "q": "is:unread"}
            )
            
            async with httpx.AsyncClient() as http:
                # List messages
                response = await http.get(token.full_url, headers=token.headers)
                data = response.json()
                
                messages = data.get("messages", [])
                if not messages:
                    print("No unread emails!")
                    return
                
                print(f"Found {len(messages)} unread emails:")
                
                # Fetch details for each message
                for msg in messages[:5]:
                    detail_token = await client.get_token(
                        "google_gmail_read_message",
                        {"id": msg["id"], "format": "metadata"}
                    )
                    detail_response = await http.get(
                        detail_token.full_url,
                        headers=detail_token.headers
                    )
                    detail = detail_response.json()
                    
                    # Extract headers
                    headers = {
                        h["name"]: h["value"] 
                        for h in detail.get("payload", {}).get("headers", [])
                    }
                    print(f"  • {headers.get('Subject', 'No subject')}")
                    print(f"    From: {headers.get('From', 'Unknown')}")
                    
    except AuthenticationError as e:
        print(f"Authentication failed: {e}")
        print("Run: afctl auth login")
    except MCPError as e:
        print(f"Token error: {e.message}")

if __name__ == "__main__":
    asyncio.run(list_unread_emails())copy

Web App with Keycloak

from fastapi import FastAPI, Depends, HTTPException
import httpx
from af_sdk import TokenClient
from af_sdk.exceptions import AuthenticationError, MCPError

app = FastAPI()

APP_ID = "org-abc123_web-app"
APP_SECRET = "sk_live_xyz789..."

async def get_token_client(keycloak_token: str):
    """Create a Token Client for the authenticated user."""
    client = TokenClient(
        method="keycloak",
        app_id=APP_ID,
        app_secret=APP_SECRET,
        keycloak_token=keycloak_token
    )
    await client.connect()
    return client

@app.get("/api/drive/files")
async def list_drive_files(
    token: str = Depends(get_keycloak_token)  # Your auth dependency
):
    """List user's Google Drive files."""
    try:
        client = await get_token_client(token)
        try:
            # Get token for Drive API
            api_token = await client.get_token("google_drive_list_files", {
                "pageSize": 20,
                "orderBy": "modifiedTime desc"
            })
            
            # Make the API call
            async with httpx.AsyncClient() as http:
                response = await http.get(
                    api_token.full_url,
                    headers=api_token.headers
                )
                return response.json()
        finally:
            await client.disconnect()
            
    except AuthenticationError:
        raise HTTPException(401, "Authentication failed")
    except MCPError as e:
        raise HTTPException(500, f"API error: {e.message}")

@app.get("/api/tools")
async def list_available_tools(
    token: str = Depends(get_keycloak_token)
):
    """List tools available to the user."""
    try:
        client = await get_token_client(token)
        try:
            await client.list_tools()
            return {"tools": client.tool_names}
        finally:
            await client.disconnect()
    except AuthenticationError:
        raise HTTPException(401, "Authentication failed")copy

Sync Script

import requests
from af_sdk import TokenClient
from af_sdk.exceptions import MCPError

def daily_calendar_check():
    """Sync script that checks today's calendar events."""
    
    with TokenClient(
        method="cli",
        app_id="org-abc123_calendar-checker",
        app_secret="sk_live_xyz789..."
    ) as client:
        # List tools (sync)
        client.list_tools_sync()
        
        if not client.has_tool("google_calendar_list_events"):
            print("Calendar not connected")
            return
        
        # Get token for Calendar API
        token = client.get_token_sync("google_calendar_list_events", {
            "calendarId": "primary",
            "timeMin": "2026-02-23T00:00:00Z",
            "timeMax": "2026-02-23T23:59:59Z",
            "singleEvents": True,
            "orderBy": "startTime"
        })
        
        # Make the API call
        response = requests.get(token.full_url, headers=token.headers)
        data = response.json()
        
        events = data.get("items", [])
        print(f"Today's events: {len(events)}")
        
        for event in events:
            start = event.get("start", {}).get("dateTime", event.get("start", {}).get("date"))
            print(f"  • {event.get('summary', 'No title')} at {start}")

if __name__ == "__main__":
    daily_calendar_check()copy

MCP Client vs Token Client

Both clients use the same authentication flow and connect to Agentic Fabriq. The difference is what they return:

Use Token Client when you need:

• Full control over HTTP requests
• Custom request/response handling
• Direct debugging of API calls
• Integration with existing HTTP infrastructure

Use MCP Client when you need:

• Simple tool execution without HTTP handling
• Standard MCP protocol compatibility
• Server-side error handling and retries