Agentic Fabriq SDK Reference

Installation

pip install agentic-fabriq-sdkcopy

Requirements: Python 3.11 – 3.12

Table of Contents

1. MCPClient
2. Authentication Functions
3. Application Helpers
4. Data Models
5. Exceptions
6. Examples

MCPClient

The main client for connecting to Agentic Fabriq and invoking tools.

Constructor

MCPClient(
    method: str,
    app_id: str,
    app_secret: str,
    keycloak_token: str = None,
    external_token: str = None,
    org_url: str = None,
    mcp_url: str = None,
    gateway_url: str = None,
    timeout: float = 60.0
)copy

Parameters

Authentication Methods

Usage

# Using context manager (recommended)
async with MCPClient(method="cli", app_id="myapp", app_secret="sk_xxx") as client:
    tools = await client.list_tools()
    result = await client.call_tool("google_gmail_get_emails", {"max_results": 10})

# Manual connection
client = MCPClient(method="cli", app_id="myapp", app_secret="sk_xxx")
await client.connect()
# ... use client
await client.disconnect()copy

Methods

connect()

Establishes connection to the MCP server. Called automatically by context manager.

await client.connect()copy

Raises:
- AuthenticationError: Invalid credentials or user not logged in
- MCPConnectionError: Cannot reach MCP server

disconnect()

Closes the connection and clears internal state.

await client.disconnect()copy

list_tools()

Fetches available tools from the MCP server.

tools = await client.list_tools()copy

Returns: List[Dict] — List of tool definitions

[
    {
        "name": "google_gmail_get_emails",
        "description": "Fetch emails from Gmail",
        "inputSchema": { ... }
    }
]copy

call_tool(name, arguments)

Invokes a tool with the given arguments.

result = await client.call_tool(name: str, arguments: dict = None)copy

Parameters:

Returns: Tool response (structure varies by tool)

Raises:
- MCPError: Tool returned an error
- MCPConnectionError: Not connected or network error

get_tools()

Returns cached tools list (no network call).

tools = client.get_tools()copy

Returns: List[Dict]

get_tool(name)

Gets a specific tool by name from cache.

tool = client.get_tool("google_gmail_get_emails")copy

Returns: Dict or None

has_tool(name)

Checks if a tool exists.

if client.has_tool("google_gmail_get_emails"):
    ...copy

Returns: bool

Properties

Sync Methods

with MCPClient(method="cli", app_id="myapp", app_secret="sk_xxx") as client:
    tools = client.list_tools_sync()copy

Authentication Functions

load_stored_credentials()

Loads credentials saved by afctl auth login from ~/.af/.

from af_sdk import load_stored_credentials

creds = load_stored_credentials()copy

Returns: StoredCredentials or None if not logged in

exchange_keycloak_for_af_token()

Exchanges a Keycloak token for an app-scoped AF token.

from af_sdk import exchange_keycloak_for_af_token

af_token = await exchange_keycloak_for_af_token(
    keycloak_token: str,
    app_id: str,
    secret_key: str,
    gateway_url: str = "https://dashboard.agenticfabriq.com"
)copy

Parameters:

Returns: AFTokenResponse

Raises: AuthenticationError

exchange_okta_for_af_token()

Exchanges an Okta SSO token for an AF token.

from af_sdk import exchange_okta_for_af_token

af_token = await exchange_okta_for_af_token(
    okta_token: str,
    app_id: str,
    app_secret: str,
    org_url: str = None,
    gateway_url: str = "https://dashboard.agenticfabriq.com"
)copy

Parameters:

Returns: str — AF access token

Raises: AuthenticationError

get_valid_token_sync()

Synchronously loads credentials and exchanges for AF token.

from af_sdk import get_valid_token_sync

token = get_valid_token_sync(
    app_id: str,
    secret_key: str,
    gateway_url: str = "https://dashboard.agenticfabriq.com"
)copy

Returns: str — AF access token

Raises: AuthenticationError — Not logged in or token expired

Application Helpers

get_application_client()

Creates an MCPClient from saved application credentials.

from af_sdk import get_application_client

client = await get_application_client(
    app_id: str,
    config_dir: Path = None,
    gateway_url: str = None
)copy

Returns: MCPClient

Raises: ApplicationNotFoundError

load_application_config()

Loads application config from ~/.af/applications/{{app_id}.json.

from af_sdk import load_application_config

config = load_application_config(app_id: str, config_dir: Path = None)copy

Returns: Dict — Application configuration

list_applications()

Lists all locally registered applications.

from af_sdk import list_applications

apps = list_applications(config_dir: Path = None)copy

Returns: List[Dict] — List of application configs

Data Models

All models are Pydantic BaseModel subclasses providing type validation and serialization.

StoredCredentials

Represents user credentials saved by afctl auth login. Stored encrypted in ~/.af/credentials.json.

from af_sdk import StoredCredentials, load_stored_credentials

creds = load_stored_credentials()
if creds and not creds.is_expired:
    print(f"User: {creds.email}")
    print(f"Expires in: {creds.expires_in} seconds")copy

Fields

Computed Properties

Example

creds = load_stored_credentials()

if creds is None:
    print("Not logged in")
elif creds.is_expired:
    print(f"Token expired {-creds.expires_in} seconds ago")
else:
    print(f"Logged in as: {creds.email}")
    print(f"User ID: {creds.user_id}")
    print(f"Organization: {creds.organization_id}")
    print(f"Token valid for: {creds.expires_in} seconds")copy

AFTokenResponse

Response from token exchange operations. Contains the app-scoped AF token that authorizes MCP requests.

from af_sdk import exchange_keycloak_for_af_token

response = await exchange_keycloak_for_af_token(
    keycloak_token="...",
    app_id="myapp",
    secret_key="sk_xxx"
)
print(f"Token: {response.access_token[:50]}...")copy

Fields

Usage

# Use token for HTTP requests
headers = {"Authorization": f"{response.token_type} {response.access_token}"}

# Check token metadata
print(f"App: {response.app_id}")
print(f"User: {response.user_id}")
print(f"Expires in: {response.expires_in}s")copy

Exceptions

All SDK exceptions inherit from AFError and follow a consistent structure for error handling.

Base Exception

AFError

Base class for all SDK exceptions.

class AFError(Exception):
    message: str          # Human-readable error description
    error_code: str       # Machine-readable error code (e.g., "SERVER_ERROR")
    request_id: str       # Unique request ID for debugging
    details: dict         # Additional error contextcopy

Constructor:

AFError(
    message: str,
    error_code: str = "SERVER_ERROR",
    request_id: str = None,
    details: dict = None
)copy

Exception Hierarchy

AFError
├── AuthenticationError     # Credentials/login issues
├── AuthorizationError      # Permission/scope issues
├── NotFoundError           # Resource doesn't exist
├── ValidationError         # Invalid input
├── MCPError                # Tool execution errors
├── MCPConnectionError      # Network/connection issues
├── ConnectorError          # Internal connector errors
├── RateLimitError          # Too many requests
├── UpstreamError           # External service errors
├── VaultError              # Secret storage errors
├── TokenRefreshError       # Token refresh failures
└── ApplicationNotFoundError # App config not foundcopy

Key Exceptions

AuthenticationError

Raised when authentication fails.

Error Code: AUTHENTICATION_FAILED

Raised when:

• User hasn't run afctl auth login (no stored credentials)
• Stored credentials have expired
• Invalid app_id or app_secret
• Token exchange with gateway fails
• Keycloak/IdP token is invalid or expired

from af_sdk import MCPClient, AuthenticationError

try:
    async with MCPClient(method="cli", app_id="myapp", app_secret="sk_xxx") as client:
        pass
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")
    print(f"Error code: {e.error_code}")  # "AUTHENTICATION_FAILED"
    print(f"Request ID: {e.request_id}")
    # Prompt user to run 'afctl auth login'copy

AuthorizationError

Raised when the user lacks permission for an operation.

Error Code: FORBIDDEN

Raised when:

• Application doesn't have required scopes for a tool
• User hasn't connected the required service (e.g., Google account not linked)
• Organization policy denies the operation
• User's role doesn't permit the action

from af_sdk import MCPClient, AuthorizationError

try:
    result = await client.call_tool("google_gmail_get_emails", {})
except AuthorizationError as e:
    print(f"Permission denied: {e.message}")
    # Check e.details for specific scope requirements
    if "required_scopes" in e.details:
        print(f"Required: {e.details['required_scopes']}")copy

MCPError

Raised when a tool returns an error during execution.

Error Code: MCP_ERROR

Additional Properties:

from af_sdk import MCPError

try:
    result = await client.call_tool("google_gmail_get_emails", {"q": "invalid:query"})
except MCPError as e:
    print(f"Tool failed: {e.message}")
    print(f"Error code: {e.code}")  # JSON-RPC code
    print(f"Request ID: {e.request_id}")
    if e.data:
        print(f"Additional info: {e.data}")copy

MCPConnectionError

Raised when connection to the MCP server fails.

Error Code: MCP_CONNECTION_ERROR

Raised when:

• Cannot establish connection to MCP server
• Connection drops during operation
• Request timeout
• DNS resolution failure
• SSL/TLS errors

from af_sdk import MCPConnectionError
import asyncio

async def call_with_retry(client, tool, args, retries=3):
    for attempt in range(retries):
        try:
            return await client.call_tool(tool, args)
        except MCPConnectionError as e:
            if attempt == retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # Exponential backoffcopy

Comprehensive Error Handling

from af_sdk import (
    MCPClient,
    AFError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    MCPError,
    MCPConnectionError,
)

async def safe_call(client, tool_name, args):
    try:
        return await client.call_tool(tool_name, args)
    
    except AuthenticationError as e:
        # User needs to log in
        return {"error": "auth_required", "message": str(e)}
    
    except AuthorizationError as e:
        # Missing permissions
        return {"error": "forbidden", "details": e.details}
    
    except NotFoundError as e:
        # Tool or resource doesn't exist
        return {"error": "not_found", "message": str(e)}
    
    except ValidationError as e:
        # Invalid parameters
        return {"error": "invalid_input", "message": str(e)}
    
    except MCPError as e:
        # Tool execution failed
        return {"error": "tool_error", "code": e.code, "request_id": e.request_id}
    
    except MCPConnectionError as e:
        # Network issue
        return {"error": "connection_failed", "message": str(e)}
    
    except AFError as e:
        # Catch-all
        return {"error": e.error_code, "message": str(e)}copy

Device Code Flow (Remote/Headless Auth)

For machines without a browser (VMs, remote servers, CI environments), the SDK supports the OAuth 2.0 Device Authorization Grant (RFC 8628). Instead of opening a browser locally, the CLI displays a URL and a code that you enter on any device with a browser.

# On the remote machine:
afctl auth login --remote

# Output:
# Visit: https://auth.agenticfabriq.com/realms/agentic-fabric/device
# Enter code: ABCD-EFGH
# Waiting for authorization...

# Open the URL on any browser, enter the code, and authorize.
# The CLI will automatically detect authorization and save your tokens.copy

This can be combined with --org for team workspace authentication on remote machines:

afctl auth login --remote --org your-domain.comcopy

Organization / Team Authentication

Organizations with dedicated Keycloak realms use the --org flag to route authentication to their realm instead of the default individual-user realm.

# Login to a team workspace
afctl auth login --org your-domain.com

# Or save the org URL so all future logins use it
afctl config set organization_url your-domain.com
afctl auth logincopy

The org URL is resolved to the correct Keycloak realm via the POST /api/v1/auth/resolve-org endpoint. Token refresh (afctl auth refresh) is realm-aware and will use the same org realm.

Coding Agent Integration (MCP Broker)

The SDK includes a local MCP broker (afctl broker) that connects coding agents — Claude Code, Cursor, and Codex — to Agentic Fabriq. The broker is spawned as a child process by the agent and proxies MCP tool calls to the gateway.

To set it up, add the following to your agent's MCP config file:

{
  "mcpServers": {
    "agentic-fabriq": {
      "command": "afctl",
      "args": ["broker"]
    }
  }
}copy

See the Coding Agents Setup guide for detailed instructions, and the CLI Reference for afctl broker options.

Examples

Basic Usage

import asyncio
from af_sdk import MCPClient

async def main():
    async with MCPClient(
        method="cli",
        app_id="org-xxx_myapp",
        app_secret="sk_xxx"
    ) as client:
        # List tools
        print(f"Tools: {client.tool_names}")
        
        # Call a tool
        result = await client.call_tool("google_gmail_get_emails", {
            "max_results": 5,
            "q": "is:unread"
        })
        print(f"Unread: {result['total_count']}")

asyncio.run(main())copy

With Keycloak Token

async with MCPClient(
    method="keycloak",
    app_id="org-xxx_myapp",
    app_secret="sk_xxx",
    keycloak_token="eyJ..."
) as client:
    result = await client.call_tool("slack_post_message", {
        "channel": "general",
        "text": "Hello!"
    })copy

With Okta SSO

async with MCPClient(
    method="idp",
    app_id="org-xxx_myapp",
    app_secret="sk_xxx",
    external_token="eyJ...",
    org_url="acme.com"
) as client:
    files = await client.call_tool("google_drive_get_files", {"max_results": 10})copy

Synchronous Usage

from af_sdk import MCPClient

with MCPClient(method="cli", app_id="myapp", app_secret="sk_xxx") as client:
    tools = client.list_tools_sync()
    result = client.call_tool_sync("google_gmail_get_emails", {"max_results": 5})copy

All Exports

from af_sdk import (
    # Client
    MCPClient,
    
    # Authentication
    load_stored_credentials,
    exchange_keycloak_for_af_token,
    exchange_okta_for_af_token,
    get_valid_token_sync,
    
    # Application Helpers
    get_application_client,
    load_application_config,
    list_applications,
    
    # Models
    StoredCredentials,
    AFTokenResponse,
    ToolInvokeRequest,
    ToolInvokeResult,
    OAuthToken,
    ErrorResponse,
    
    # Exceptions
    AFError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    MCPError,
    MCPConnectionError,
    ConnectorError,
    RateLimitError,
    UpstreamError,
    VaultError,
    TokenRefreshError,
    ApplicationNotFoundError,
)copy

Environment Variables