Home DocsHow to Build an Agent with AF

How to Build an Agent with AF

A code-first walkthrough for writing an agent that connects to an MCP server, lists tools, and calls them. Pasteable into a coding agent as the source of truth — every code block is runnable once the env vars below are set.

The agent code is the same whether you're running on a laptop or shipping to production. The differences live at the edges: how secrets reach the process, how the runtime is shaped, how users sign in. The first part of this guide covers writing the agent. The last section (Deploying to production) covers secret-loading and containerization patterns.

Shortcut: don't want to write this by hand? Drop the General Agent Prompt into Claude Code, Cursor, or Codex and your coding agent builds this integration for you — this guide is what it follows.

TL;DR (How to call AF's MCP Server)

pip install agentic-fabriq-sdk
export AF_APP_ID='...'
export AF_APP_SECRET='...'
export AF_GATEWAY_URL='...'

# agent.py
import asyncio, os
from af_sdk import MCPClient

async def main():
    async with MCPClient(
        method="cli",                            # see auth methods below
        app_id=os.environ["AF_APP_ID"],
        app_secret=os.environ["AF_APP_SECRET"],
        gateway_url=os.environ["AF_GATEWAY_URL"],
    ) as client:
        tools = await client.list_tools()
        print(f"{len(tools)} tools available")
        result = await client.call_tool(
            "google_gmail_list_messages",
            {"max_results": 5},
        )
        print(result)

asyncio.run(main())

python agent.py

What you need before you write code
Install
The MCPClient — four ways to authenticate
Listing and calling tools
Loops, retries, and error handling
Sync (non-async) agents
Full reference agent
Deploying to production
- Loading secrets
- Containerized reference agent

What you need before you write code

Item	Required	Notes
`AF_APP_ID`	yes	The app's client ID.
`AF_APP_SECRET`	yes	The app's client secret. Treat as a password.
`AF_GATEWAY_URL`	yes	Where the AF gateway is reachable.
`AF_MCP_URL`	optional	MCP server URL. If unset, the SDK derives it from `AF_GATEWAY_URL`.
`AF_KEYCLOAK_URL`	Method B	Where Keycloak is reachable.
`KC_REALM`	Method B	The Keycloak realm name your account lives in.
Keycloak access token	Method B	A raw OIDC token from your Keycloak realm.
Okta (or other IdP) ID token	Method C	A JWT from the SSO provider configured for your app.
AF JWT minted by your backend	Method D	Used for B2B2C — your backend fetches a per-end-user AF token.

Don't hardcode URLs in code. Read them from env vars and decide per environment.

Install

pip install agentic-fabriq-sdk

The SDK exposes everything an agent needs from the top level:

from af_sdk import (
    MCPClient,
    AuthenticationError,
    MCPError,
    MCPConnectionError,
)

The `MCPClient` — four ways to authenticate

MCPClient takes method=... to choose how it acquires an AF JWT. The agent's app_id + app_secret are required in every method — they identify the application. The optional token argument identifies the human or external user the application is acting as.

MCPClient(
    *,
    method: Literal["cli", "keycloak", "idp", "token"],
    app_id: str,
    app_secret: str,
    keycloak_token: str | None = None,        # method="keycloak"
    keycloak_refresh_token: str | None = None,# method="keycloak", optional
    external_token: str | None = None,        # method="idp"
    af_token: str | None = None,              # method="token"
    org_url: str | None = None,
    mcp_url: str | None = None,
    gateway_url: str | None = None,
    timeout: float = 60.0,
    auto_refresh: bool = True,                # method="cli", true if you want auto refresh of tokens on
)

async with MCPClient(...) as client: is the idiomatic usage. The context manager calls connect() on entry (which exchanges credentials and fetches the tool list) and disconnect() on exit.

Method A — `cli` (developer session via `afctl`)

Use when you're the human running the agent and you've already authenticated with afctl auth login. The SDK reads stored credentials from ~/.af/tokens.enc, refreshes them if needed, then exchanges them for an AF token.

afctl auth login --keycloak-url "$AF_KEYCLOAK_URL"

# agent_cli_auth.py
import asyncio, os
from af_sdk import MCPClient

async def main():
    async with MCPClient(
        method="cli",
        app_id=os.environ["AF_APP_ID"],
        app_secret=os.environ["AF_APP_SECRET"],
        gateway_url=os.environ["AF_GATEWAY_URL"],
    ) as client:
        for tool in await client.list_tools():
            print(tool["name"])

asyncio.run(main())

When to use: development, scripts you run from your own laptop, demos. Not appropriate for headless servers, CI, or production agents — the agent shouldn't depend on an interactive login.

Method B — `keycloak` (raw Keycloak access token)

Use when a human signs into your frontend, your backend exchanges the OIDC ?code=... for a token pair, and the agent uses that token to act on the user's behalf.

Configure the Keycloak client

You don't have to do this yourself — when AF registers an application, it provisions the underlying Keycloak OIDC client with the right settings for the authorization_code flow. For reference, the resulting client looks like this:

Setting	Value
Client type	`OpenID Connect`, public (PKCE) or confidential (backend-channel)
`Client authentication`	On for confidential, Off for public
`Standard flow` (authorization_code)	Enabled
`Direct access grants` (password)	Disabled
Valid redirect URIs	Wildcard registered by AF — see the redirect-URI section below
Web origins	Wildcard for CORS during PKCE
Client ID / secret	Client ID = the AF `app_id`; secret = the AF `app_secret` (for confidential clients)

The realm name goes into every Keycloak URL as realms/<KC_REALM>/....

Browser-based (`authorization_code`)

The user clicks "Log in" → browser redirects to Keycloak → user authenticates → Keycloak redirects back to your callback URL with ?code=... → your backend exchanges the code for tokens → the agent runs with those tokens.

Frontend — kick off the login

PKCE flow (recommended for SPAs; client secret never leaves the backend):

// login.ts (frontend)
async function startLogin() {
  const codeVerifier  = base64UrlEncode(crypto.getRandomValues(new Uint8Array(64)));
  const codeChallenge = base64UrlEncode(await sha256(codeVerifier));
  sessionStorage.setItem("kc_code_verifier", codeVerifier);

  const params = new URLSearchParams({
    response_type:         "code",
    client_id:             import.meta.env.VITE_KC_CLIENT_ID,
    redirect_uri:          `${import.meta.env.VITE_APP_ORIGIN}/oauth/callback`,
    scope:                 "openid profile email",
    code_challenge:        codeChallenge,
    code_challenge_method: "S256",
    state:                 crypto.randomUUID(),
  });

  window.location.assign(
    `${import.meta.env.VITE_AF_KEYCLOAK_URL}/realms/${import.meta.env.VITE_KC_REALM}/protocol/openid-connect/auth?${params}`
  );
}

After login, the browser lands at /oauth/callback?code=...&state=.... The frontend forwards the code (and code_verifier) to the backend.

Backend — exchange the code for tokens

# fastapi_kc_callback.py
from fastapi import FastAPI, Request, HTTPException
from af_sdk import MCPClient
import os, httpx

app = FastAPI()

async def exchange_code(code: str, code_verifier: str) -> dict:
    async with httpx.AsyncClient() as http:
        resp = await http.post(
            f"{os.environ['AF_KEYCLOAK_URL']}/realms/{os.environ['KC_REALM']}/protocol/openid-connect/token",
            data={
                "grant_type":    "authorization_code",
                "code":          code,
                "redirect_uri":  os.environ["KC_REDIRECT_URI"],
                "client_id":     os.environ["KC_CLIENT_ID"],
                "client_secret": os.environ.get("KC_CLIENT_SECRET", ""),  # public client → empty
                "code_verifier": code_verifier,
            },
        )
        resp.raise_for_status()
        return resp.json()

@app.post("/oauth/callback")
async def callback(payload: dict, request: Request):
    tokens = await exchange_code(payload["code"], payload["code_verifier"])
    # Store both tokens in a server-side session, NOT in localStorage.
    request.session["kc_access_token"]  = tokens["access_token"]
    request.session["kc_refresh_token"] = tokens["refresh_token"]
    return {"ok": True}

@app.post("/run-agent")
async def run_agent(request: Request, prompt: str):
    async with MCPClient(
        method="keycloak",
        app_id=os.environ["AF_APP_ID"],
        app_secret=os.environ["AF_APP_SECRET"],
        keycloak_token=request.session["kc_access_token"],
        keycloak_refresh_token=request.session["kc_refresh_token"],
        gateway_url=os.environ["AF_GATEWAY_URL"],
    ) as client:
        return await client.call_tool("google_gmail_send_message", {"body": prompt})

Checklist for the callback URL:

Keycloak won't enforce a redirect URI for you. AF registers a wildcard Valid redirect URI on the Keycloak client, so any redirect_uri your frontend sends will be accepted by Keycloak. That means your code is the only thing keeping the handshake correct — the platform won't catch a mismatch for you.
Be precise with yourself. The redirect_uri your frontend sends in the /auth request must be byte-identical to the redirect_uri your backend sends in the code-exchange request. Mismatched scheme, host, port, path, or even trailing slash will cause the exchange to fail — not because Keycloak rejects it, but because the OAuth spec requires the values to match across the two calls.
A common shape: pick one redirect_uri per environment (http://localhost:3000/oauth/callback for dev, https://app.example.com/oauth/callback for prod). Drive both the frontend and the backend from the same env var so they can't drift.
Store tokens in an httpOnly session cookie or your server-side session store. Never put them in localStorage or sessionStorage long-term.
Pass state on login and validate it on callback — required to defend against CSRF.

When to use: anything where a human user signs in through your UI and the agent operates on their behalf.

Method C — `idp` (Okta / external SSO token)

When the user signs in with an external IdP (Okta is the supported case), your frontend gets an Okta ID token. The SDK exchanges it for an AF JWT.

Configure the Okta application

Setting	Value
Application type	OIDC — Web application (or Single-Page Application for PKCE)
Grant types	Authorization Code (+ Refresh Token if you want renewal)
Sign-in redirect URIs	`https://<your-frontend>/oauth/okta/callback` — one entry per env
Sign-out redirect URIs	Optional but recommended
Trusted origins	`https://<your-frontend>` (Web Origins for CORS)
Client ID / secret	Store as `OKTA_CLIENT_ID` / `OKTA_CLIENT_SECRET`
Issuer URL	Found under "Sign On" in the Okta app; store as `OKTA_ISSUER`
Scopes	`openid profile email` plus any scopes the AF app expects

Whatever redirect URI you set in Okta must match exactly the redirect_uri your frontend sends — including scheme, host, port, and path.

Frontend — kick off the Okta login

// okta_login.ts (frontend)
async function startOktaLogin() {
  const codeVerifier  = base64UrlEncode(crypto.getRandomValues(new Uint8Array(64)));
  const codeChallenge = base64UrlEncode(await sha256(codeVerifier));
  sessionStorage.setItem("okta_code_verifier", codeVerifier);

  const params = new URLSearchParams({
    response_type:         "code",
    client_id:             import.meta.env.VITE_OKTA_CLIENT_ID,
    redirect_uri:          `${import.meta.env.VITE_APP_ORIGIN}/oauth/okta/callback`,
    scope:                 "openid profile email",
    code_challenge:        codeChallenge,
    code_challenge_method: "S256",
    state:                 crypto.randomUUID(),
  });

  window.location.assign(
    `${import.meta.env.VITE_OKTA_ISSUER}/v1/authorize?${params}`
  );
}

After Okta authenticates the user, the browser lands at /oauth/okta/callback?code=...&state=.... The frontend forwards code (and code_verifier) to the backend.

Backend — exchange the code, then run the agent

# fastapi_okta_callback.py
from fastapi import FastAPI, Request
from af_sdk import MCPClient
import os, httpx

app = FastAPI()

async def exchange_okta_code(code: str, code_verifier: str) -> dict:
    async with httpx.AsyncClient() as http:
        resp = await http.post(
            f"{os.environ['OKTA_ISSUER']}/v1/token",
            data={
                "grant_type":    "authorization_code",
                "code":          code,
                "redirect_uri":  os.environ["OKTA_REDIRECT_URI"],
                "client_id":     os.environ["OKTA_CLIENT_ID"],
                "client_secret": os.environ.get("OKTA_CLIENT_SECRET", ""),  # SPA → empty
                "code_verifier": code_verifier,
            },
        )
        resp.raise_for_status()
        return resp.json()

@app.post("/oauth/okta/callback")
async def okta_callback(payload: dict, request: Request):
    tokens = await exchange_okta_code(payload["code"], payload["code_verifier"])
    request.session["okta_id_token"] = tokens["id_token"]
    return {"ok": True}

@app.post("/run-agent")
async def run_agent(request: Request, prompt: str):
    async with MCPClient(
        method="idp",
        app_id=os.environ["AF_APP_ID"],
        app_secret=os.environ["AF_APP_SECRET"],
        external_token=request.session["okta_id_token"],
        org_url=os.environ.get("ORG_URL"),                # optional, for org-realm routing
        gateway_url=os.environ["AF_GATEWAY_URL"],
    ) as client:
        return await client.call_tool("google_gmail_send_message", {"body": prompt})

Callback-URL checklist (identical to the Keycloak case but with Okta as the IdP):

The redirect_uri you send during login must exactly match one of the "Sign-in redirect URIs" you registered in the Okta app.
One redirect URI per environment, all registered.
Validate state on callback.
Store id_token server-side (httpOnly session / Redis), not in localStorage.

When to use: enterprise deployments where users sign in via Okta SSO and the agent must act as that user.

Method D — `token` (pre-exchanged AF token, e.g. B2B2C)

Use when your backend has already minted an AF token for a specific end user (typical in B2B2C: your service holds the app secret, your end users don't have AF accounts, and your backend mints per-user tokens via the external-user APIs).

# agent_b2b2c.py
import asyncio, os, httpx
from af_sdk import MCPClient

async def fetch_af_token_for_external_user(external_user_id: str) -> str:
    """Your backend hits AF's external-user token endpoint with the app secret."""
    gateway = os.environ["AF_GATEWAY_URL"]
    async with httpx.AsyncClient() as http:
        # First, get an app-level service token.
        token_resp = await http.post(
            f"{gateway}/api/v1/applications/token",
            json={
                "app_id":     os.environ["AF_APP_ID"],
                "secret_key": os.environ["AF_APP_SECRET"],
            },
        )
        token_resp.raise_for_status()
        app_token = token_resp.json()["access_token"]

        # Now mint a per-external-user token.
        user_resp = await http.post(
            f"{gateway}/api/v1/apps/{os.environ['AF_APP_ID']}/external-users/{external_user_id}/token",
            headers={"Authorization": f"Bearer {app_token}"},
        )
        user_resp.raise_for_status()
        return user_resp.json()["access_token"]

async def main():
    af_token = await fetch_af_token_for_external_user(os.environ["EXTERNAL_USER_ID"])

    async with MCPClient(
        method="token",
        app_id=os.environ["AF_APP_ID"],
        app_secret=os.environ["AF_APP_SECRET"],
        af_token=af_token,
        gateway_url=os.environ["AF_GATEWAY_URL"],
    ) as client:
        result = await client.call_tool(
            "google_gmail_send_message",
            {"to": os.environ["TO_EMAIL"], "subject": "hi", "body": "from agent"},
        )
        print(result)

asyncio.run(main())

Cache the AF token per (external_user_id, expiry) in Redis or your session store. Don't refetch on every request.

When to use: B2B2C — your product has its own users, AF is invisible to them, and your backend mints AF tokens per end user.

Listing and calling tools

Once MCPClient.connect() has run (which happens automatically inside async with), the tool list is cached on the client:

async with MCPClient(...) as client:
    # All three are equivalent — get_tools() is cached, list_tools() refetches.
    tools = client.get_tools()
    tools = await client.list_tools()
    print(client.tool_names)        # ['google_gmail_list_messages', 'slack_post_message', ...]

    # Look one up:
    tool = client.get_tool("google_gmail_list_messages")
    if tool:
        print(tool["description"])
        print(tool["inputSchema"])

    # Call it:
    messages = await client.call_tool(
        "google_gmail_list_messages",
        {"max_results": 5, "query": "is:unread"},
    )
    for m in messages:
        print(m.get("subject"))

call_tool returns whatever the MCP server gives back — usually a list of content items or a structured dict. Check the tool's inputSchema (JSON Schema) for required arguments.

Loops, retries, and error handling

Agents are usually long-lived — they tick on a schedule, react to webhooks, or run an LLM-driven loop. Three things tend to go wrong: tokens expire, the network blips, a tool call fails.

# agent_loop.py
import asyncio, logging, os
from af_sdk import MCPClient, AuthenticationError, MCPError, MCPConnectionError

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("agent")

async def call_with_retry(client: MCPClient, name: str, args: dict, attempts: int):
    for i in range(1, attempts + 1):
        try:
            return await client.call_tool(name, args)
        except MCPError as e:
            log.warning("%s failed (attempt %d/%d): %s", name, i, attempts, e)
            if i == attempts:
                raise
            await asyncio.sleep(2 ** i)

async def tick(client: MCPClient):
    unread = await call_with_retry(
        client, "google_gmail_list_messages",
        {"max_results": 10, "query": "is:unread"},
        attempts=int(os.environ["TOOL_RETRY_ATTEMPTS"]),
    )
    log.info("found %d unread", len(unread))

async def main():
    while True:
        try:
            async with MCPClient(
                method="keycloak",
                app_id=os.environ["AF_APP_ID"],
                app_secret=os.environ["AF_APP_SECRET"],
                keycloak_token=os.environ["KC_ACCESS_TOKEN"],
                keycloak_refresh_token=os.environ.get("KC_REFRESH_TOKEN"),
                gateway_url=os.environ["AF_GATEWAY_URL"],
            ) as client:
                while True:
                    await tick(client)
                    await asyncio.sleep(int(os.environ["TICK_SECONDS"]))
        except AuthenticationError:
            log.exception("auth failed — refresh credentials and retry")
            await asyncio.sleep(int(os.environ["AUTH_BACKOFF_SECONDS"]))
        except MCPConnectionError:
            log.exception("connection lost — reconnecting")
            await asyncio.sleep(int(os.environ["RECONNECT_BACKOFF_SECONDS"]))

asyncio.run(main())

Key behaviors:

auto_refresh=True (default) renews the AF token in-place when it's about to expire, provided you passed a refresh token (Method B) or the source-of-truth token can be re-acquired (Methods A/C/D depend on caller).
AuthenticationError is raised on initial auth failure; catch it at the outermost loop and re-fetch credentials.
MCPConnectionError covers transport-level failures; re-entering the async with block reconnects.
MCPError covers tool-call failures (bad arguments, downstream API errors); usually retryable for transient failures, not for 4xx-class errors.

Sync (non-async) agents

If you can't run an event loop (e.g., you're embedding in a sync framework), use with instead of async with and the _sync method variants:

# sync_agent.py
import os
from af_sdk import MCPClient

with MCPClient(
    method="cli",
    app_id=os.environ["AF_APP_ID"],
    app_secret=os.environ["AF_APP_SECRET"],
    gateway_url=os.environ["AF_GATEWAY_URL"],
) as client:
    tools = client.list_tools_sync()
    result = client.call_tool_sync("google_gmail_list_messages", {"max_results": 5})
    print(result)

The sync wrappers spin up a private event loop per call. Don't use them inside an already-running async context.

Full reference agent

A complete, opinionated reference. Auto-picks an auth method based on which env vars are present, logs structured JSON, retries transient errors, runs as a configurable tick loop. Drop into a file, set env vars, run.

# reference_agent.py
import asyncio, logging, os, sys
from typing import Any
from af_sdk import (
    MCPClient,
    AuthenticationError,
    MCPError,
    MCPConnectionError,
)

logging.basicConfig(
    level=os.environ["LOG_LEVEL"],
    format='{"ts":"%(asctime)s","lvl":"%(levelname)s","msg":"%(message)s"}',
)
log = logging.getLogger("agent")

APP_ID      = os.environ["AF_APP_ID"]
APP_SECRET  = os.environ["AF_APP_SECRET"]
GATEWAY_URL = os.environ["AF_GATEWAY_URL"]

def build_client() -> MCPClient:
    """Pick auth method by which env vars are present."""
    if "KC_ACCESS_TOKEN" in os.environ:
        return MCPClient(
            method="keycloak",
            app_id=APP_ID, app_secret=APP_SECRET,
            keycloak_token=os.environ["KC_ACCESS_TOKEN"],
            keycloak_refresh_token=os.environ.get("KC_REFRESH_TOKEN"),
            gateway_url=GATEWAY_URL,
        )
    if "OKTA_ID_TOKEN" in os.environ:
        return MCPClient(
            method="idp",
            app_id=APP_ID, app_secret=APP_SECRET,
            external_token=os.environ["OKTA_ID_TOKEN"],
            org_url=os.environ.get("ORG_URL"),
            gateway_url=GATEWAY_URL,
        )
    if "AF_USER_TOKEN" in os.environ:
        return MCPClient(
            method="token",
            app_id=APP_ID, app_secret=APP_SECRET,
            af_token=os.environ["AF_USER_TOKEN"],
            gateway_url=GATEWAY_URL,
        )
    # Developer fallback — requires `afctl auth login`.
    return MCPClient(
        method="cli",
        app_id=APP_ID, app_secret=APP_SECRET,
        gateway_url=GATEWAY_URL,
    )

async def call_with_retry(client: MCPClient, name: str, args: dict[str, Any], attempts: int) -> Any:
    for i in range(1, attempts + 1):
        try:
            return await client.call_tool(name, args)
        except MCPError as e:
            log.warning("tool %s failed (%d/%d): %s", name, i, attempts, e)
            if i == attempts:
                raise
            await asyncio.sleep(min(2 ** i, int(os.environ["MAX_TOOL_BACKOFF_SECONDS"])))

async def tick(client: MCPClient) -> None:
    """One unit of agent work. Replace with your business logic."""
    if not client.has_tool("google_gmail_list_messages"):
        log.warning("gmail tool not available; skipping tick")
        return
    msgs = await call_with_retry(
        client, "google_gmail_list_messages",
        {"max_results": 10, "query": "is:unread"},
        attempts=int(os.environ["TOOL_RETRY_ATTEMPTS"]),
    )
    log.info("processed %d unread messages", len(msgs or []))

async def run_forever() -> None:
    backoff = int(os.environ["INITIAL_BACKOFF_SECONDS"])
    max_backoff = int(os.environ["MAX_BACKOFF_SECONDS"])
    while True:
        try:
            async with build_client() as client:
                log.info("connected; %d tools available", len(client.get_tools()))
                backoff = int(os.environ["INITIAL_BACKOFF_SECONDS"])
                while True:
                    await tick(client)
                    await asyncio.sleep(int(os.environ["TICK_SECONDS"]))
        except AuthenticationError:
            log.exception("auth failed; retrying in %ds", backoff)
        except MCPConnectionError:
            log.exception("connection lost; retrying in %ds", backoff)
        except Exception:
            log.exception("unexpected error; retrying in %ds", backoff)
        await asyncio.sleep(backoff)
        backoff = min(backoff * 2, max_backoff)

if __name__ == "__main__":
    try:
        asyncio.run(run_forever())
    except KeyboardInterrupt:
        log.info("shutdown")
        sys.exit(0)

Run it:

AF_APP_ID='...' \
AF_APP_SECRET='...' \
AF_GATEWAY_URL='...' \
KC_ACCESS_TOKEN='...' \
LOG_LEVEL='INFO' \
TICK_SECONDS='60' \
TOOL_RETRY_ATTEMPTS='3' \
MAX_TOOL_BACKOFF_SECONDS='30' \
INITIAL_BACKOFF_SECONDS='5' \
MAX_BACKOFF_SECONDS='300' \
python reference_agent.py

Deploying to production

The agent code above doesn't change between local and prod. What changes is how secrets reach the process and how the process is shaped.

Loading secrets

Don't bake AF_APP_ID / AF_APP_SECRET (or Keycloak / Okta secrets) into images, env files committed to git, or .env shipped to prod. Use your platform's secret manager.

GCP Secret Manager

# secrets_gcp.py
from google.cloud import secretmanager
import os

def load(secret_name: str, project: str) -> str:
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{project}/secrets/{secret_name}/versions/latest"
    return client.access_secret_version(name=name).payload.data.decode()

project = os.environ["GCP_PROJECT"]
os.environ["AF_APP_ID"]     = load(os.environ["AF_APP_ID_SECRET_NAME"],     project)
os.environ["AF_APP_SECRET"] = load(os.environ["AF_APP_SECRET_SECRET_NAME"], project)

AWS Secrets Manager

# secrets_aws.py
import boto3, json, os

def load(secret_id: str) -> dict:
    client = boto3.client("secretsmanager", region_name=os.environ["AWS_REGION"])
    payload = client.get_secret_value(SecretId=secret_id)["SecretString"]
    return json.loads(payload)

creds = load(os.environ["AF_SECRET_ID"])
os.environ["AF_APP_ID"]     = creds["app_id"]
os.environ["AF_APP_SECRET"] = creds["app_secret"]

HashiCorp Vault

# secrets_vault.py
import hvac, os

vault = hvac.Client(url=os.environ["VAULT_ADDR"], token=os.environ["VAULT_TOKEN"])
data = vault.secrets.kv.v2.read_secret_version(path=os.environ["VAULT_PATH"])["data"]["data"]
os.environ["AF_APP_ID"]     = data["app_id"]
os.environ["AF_APP_SECRET"] = data["app_secret"]

Kubernetes secret mounted as env vars

# k8s-deployment.yaml (excerpt)
env:
  - name: AF_APP_ID
    valueFrom:
      secretKeyRef:
        name: af-agent
        key: app_id
  - name: AF_APP_SECRET
    valueFrom:
      secretKeyRef:
        name: af-agent
        key: app_secret
  - name: AF_GATEWAY_URL
    value: ""   # set per environment

After secrets are loaded into env, the agent code is identical to the development version.

Containerized reference agent

Same reference_agent.py from the previous section, packaged for production: graceful shutdown on SIGTERM, structured logs, env-driven auth.

# agent/main.py
import asyncio, logging, os, signal, sys
from typing import Any
from af_sdk import (
    MCPClient,
    AuthenticationError,
    MCPError,
    MCPConnectionError,
)

logging.basicConfig(
    level=os.environ["LOG_LEVEL"],
    format='{"ts":"%(asctime)s","lvl":"%(levelname)s","msg":"%(message)s"}',
)
log = logging.getLogger("agent")

APP_ID      = os.environ["AF_APP_ID"]
APP_SECRET  = os.environ["AF_APP_SECRET"]
GATEWAY_URL = os.environ["AF_GATEWAY_URL"]

def build_client() -> MCPClient:
    if "KC_ACCESS_TOKEN" in os.environ:
        return MCPClient(
            method="keycloak",
            app_id=APP_ID, app_secret=APP_SECRET,
            keycloak_token=os.environ["KC_ACCESS_TOKEN"],
            keycloak_refresh_token=os.environ.get("KC_REFRESH_TOKEN"),
            gateway_url=GATEWAY_URL,
        )
    if "OKTA_ID_TOKEN" in os.environ:
        return MCPClient(
            method="idp",
            app_id=APP_ID, app_secret=APP_SECRET,
            external_token=os.environ["OKTA_ID_TOKEN"],
            org_url=os.environ.get("ORG_URL"),
            gateway_url=GATEWAY_URL,
        )
    if "AF_USER_TOKEN" in os.environ:
        return MCPClient(
            method="token",
            app_id=APP_ID, app_secret=APP_SECRET,
            af_token=os.environ["AF_USER_TOKEN"],
            gateway_url=GATEWAY_URL,
        )
    return MCPClient(
        method="cli",
        app_id=APP_ID, app_secret=APP_SECRET,
        gateway_url=GATEWAY_URL,
    )

async def call_with_retry(client, name, args, attempts) -> Any:
    for i in range(1, attempts + 1):
        try:
            return await client.call_tool(name, args)
        except MCPError as e:
            log.warning("tool %s failed (%d/%d): %s", name, i, attempts, e)
            if i == attempts:
                raise
            await asyncio.sleep(min(2 ** i, int(os.environ["MAX_TOOL_BACKOFF_SECONDS"])))

async def tick(client: MCPClient) -> None:
    """Your business logic. Replace with whatever the agent does."""
    if not client.has_tool("google_gmail_list_messages"):
        log.warning("required tool missing; skipping tick")
        return
    msgs = await call_with_retry(
        client, "google_gmail_list_messages",
        {"max_results": 10, "query": "is:unread"},
        attempts=int(os.environ["TOOL_RETRY_ATTEMPTS"]),
    )
    log.info("processed %d unread", len(msgs or []))

async def run_forever(shutdown: asyncio.Event) -> None:
    backoff = int(os.environ["INITIAL_BACKOFF_SECONDS"])
    max_backoff = int(os.environ["MAX_BACKOFF_SECONDS"])
    while not shutdown.is_set():
        try:
            async with build_client() as client:
                log.info("connected; %d tools", len(client.get_tools()))
                backoff = int(os.environ["INITIAL_BACKOFF_SECONDS"])
                while not shutdown.is_set():
                    await tick(client)
                    try:
                        await asyncio.wait_for(
                            shutdown.wait(),
                            timeout=int(os.environ["TICK_SECONDS"]),
                        )
                    except asyncio.TimeoutError:
                        pass
        except AuthenticationError:
            log.exception("auth failed; retry in %ds", backoff)
        except MCPConnectionError:
            log.exception("connection lost; retry in %ds", backoff)
        except Exception:
            log.exception("unexpected error; retry in %ds", backoff)
        try:
            await asyncio.wait_for(shutdown.wait(), timeout=backoff)
        except asyncio.TimeoutError:
            pass
        backoff = min(backoff * 2, max_backoff)

def main() -> None:
    loop = asyncio.new_event_loop()
    shutdown = asyncio.Event()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, shutdown.set)
    try:
        loop.run_until_complete(run_forever(shutdown))
    finally:
        loop.close()
        log.info("shutdown complete")

if __name__ == "__main__":
    main()
    sys.exit(0)

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
RUN pip install --no-cache-dir agentic-fabriq-sdk
COPY agent/ /app/agent/
ENV PYTHONUNBUFFERED=1
CMD ["python", "-m", "agent.main"]

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: af-agent
spec:
  replicas: 1
  selector:
    matchLabels: {app: af-agent}
  template:
    metadata:
      labels: {app: af-agent}
    spec:
      containers:
      - name: agent
        image: ""    # set per environment
        env:
          - name: AF_GATEWAY_URL
            value: ""    # set per environment
          - name: AF_APP_ID
            valueFrom: {secretKeyRef: {name: af-agent, key: app_id}}
          - name: AF_APP_SECRET
            valueFrom: {secretKeyRef: {name: af-agent, key: app_secret}}
          - name: KC_CLIENT_ID
            valueFrom: {secretKeyRef: {name: af-agent, key: kc_client_id}}
          - name: KC_CLIENT_SECRET
            valueFrom: {secretKeyRef: {name: af-agent, key: kc_client_secret}}
          - name: AF_KEYCLOAK_URL
            value: ""    # set per environment
          - name: KC_REALM
            value: ""    # set per environment
          - name: LOG_LEVEL
            value: ""    # set per environment
          - name: TICK_SECONDS
            value: ""    # set per environment
          - name: TOOL_RETRY_ATTEMPTS
            value: ""    # set per environment
          - name: MAX_TOOL_BACKOFF_SECONDS
            value: ""    # set per environment
          - name: INITIAL_BACKOFF_SECONDS
            value: ""    # set per environment
          - name: MAX_BACKOFF_SECONDS
            value: ""    # set per environment
        resources:
          requests: {cpu: "", memory: ""}
          limits:   {cpu: "", memory: ""}

Need help?

Our team is here to help you get started.

Book a call Contact us