Skip to content

Code Examples

Working examples for integrating with the Fermi API.

Getting Started

Step 1: Create an API Key

Create an API key using user authentication. See API Keys for details.

curl -X POST https://api.fermi.dev/public/v1/identity/api-keys \
  -H "Authorization: Bearer <your-user-jwt-token>" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "My First API Key",
    "scopes": ["provision:service:create", "provision:token:exchange"]
  }'

Response: 

{
  "id": "507f1f77bcf86cd799439011",
  "name": "My First API Key",
  "scopes": ["provision:service:create", "provision:token:exchange"],
  "apiKey": "fmk_live_abc123...",
  "createdAt": "2025-01-01T00:00:00Z"
}

Step 2: Create a Service Actor

curl -X POST https://api.fermi.dev/public/v1/identity/service-actors \
  -H "Authorization: Bearer fmk_live_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "My Service Actor",
    "organisationId": "your-org-id",
    "capabilities": ["analytics:query"]
  }'

Step 3: Exchange for Token

curl -X POST https://api.fermi.dev/public/v1/identity/auth/token/exchange \
  -H "Authorization: Bearer fmk_live_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "serviceActorId": "service-actor-id"
  }'

Step 4: Use Token for API Calls

import requests

token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

response = requests.post(
    "https://api.fermi.dev/agents-service/api/v1/agentcore/chat",
    headers={
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    },
    json={"message": "What were last month sales?", "session_id": "my-chat-session-123"}
)

if response.status_code == 200:
    data = response.json()
    print(data)
else:
    print(f"Error: {response.status_code}")

Complete Integration Example

Python Example

import os
import requests

API_KEY = os.getenv("FERMI_API_KEY")
SERVICE_ACTOR_ID = os.getenv("FERMI_SERVICE_ACTOR_ID")

def exchange_token(api_key, service_actor_id):
    """Exchange API key for service actor token"""
    response = requests.post(
        "https://api.fermi.dev/public/v1/identity/auth/token/exchange",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json={"serviceActorId": service_actor_id}
    )
    response.raise_for_status()
    return response.json()

def send_chat(token, query):
    """Send an Ask Fermi message using a service actor token"""
    response = requests.post(
        "https://api.fermi.dev/agents-service/api/v1/agentcore/chat",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        },
        json={"message": query, "session_id": "my-chat-session-123"}
    )
    response.raise_for_status()
    return response.json()

# Main flow
try:
    # Exchange for token
    token_data = exchange_token(API_KEY, SERVICE_ACTOR_ID)
    token = token_data["token"]

    # Send chat
    result = send_chat(token, "What were last month sales?")
    print(result)

except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        print("Token expired - exchange for a new token")
    else:
        print(f"Error: {e.response.status_code}")

JavaScript/TypeScript Example

const API_KEY = process.env.FERMI_API_KEY;
const SERVICE_ACTOR_ID = process.env.FERMI_SERVICE_ACTOR_ID;

async function exchangeToken(apiKey, serviceActorId) {
  const response = await fetch('https://api.fermi.dev/public/v1/identity/auth/token/exchange', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${apiKey}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ serviceActorId })
  });

  if (!response.ok) {
    throw new Error(`Token exchange failed: ${response.status}`);
  }

  return await response.json();
}

async function sendChat(token, query) {
  const response = await fetch('https://api.fermi.dev/agents-service/api/v1/agentcore/chat', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${token}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ message: query, session_id: 'my-chat-session-123' })
  });

  if (!response.ok) {
    throw new Error(`Chat request failed: ${response.status}`);
  }

  return await response.json();
}

// Main flow
(async () => {
  try {
    const tokenData = await exchangeToken(API_KEY, SERVICE_ACTOR_ID);
    const result = await sendChat(tokenData.token, 'What were last month sales?');
    console.log(result);
  } catch (error) {
    console.error('Error:', error.message);
  }
})();

Error Handling

Handle Token Expiration

import requests

def make_request_with_token_refresh(api_key, service_actor_id, endpoint, data):
    """Make request with automatic token refresh"""
    # Get token
    token_response = requests.post(
        "https://api.fermi.dev/public/v1/identity/auth/token/exchange",
        headers={"Authorization": f"Bearer {api_key}"},
        json={"serviceActorId": service_actor_id}
    )
    token_response.raise_for_status()
    token = token_response.json()["token"]

    # Make request
    response = requests.post(
        endpoint,
        headers={"Authorization": f"Bearer {token}"},
        json=data
    )

    # If token expired, refresh and retry
    if response.status_code == 401:
        token_response = requests.post(
            "https://api.fermi.dev/public/v1/identity/auth/token/exchange",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"serviceActorId": service_actor_id}
        )
        token = token_response.json()["token"]
        response = requests.post(
            endpoint,
            headers={"Authorization": f"Bearer {token}"},
            json=data
        )

    response.raise_for_status()
    return response.json()

Handle Rate Limits

import time
import requests

def make_request_with_backoff(url, headers, data=None, max_retries=3):
    """Make request with exponential backoff for rate limits"""
    for attempt in range(max_retries):
        if data:
            response = requests.post(url, headers=headers, json=data)
        else:
            response = requests.get(url, headers=headers)

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            wait_time = retry_after * (2 ** attempt)
            time.sleep(wait_time)
            continue

        response.raise_for_status()
        return response.json()

    raise Exception("Max retries exceeded")