Common Patterns

Production-ready Classic API patterns for conversation lifecycle, user models, webhooks, routing, and CI simulation gates.

Production-ready patterns for building with the Amigo AI API. Each pattern includes runnable Python and TypeScript examples.

Classic API. These patterns use the Classic API SDKs and endpoints at api.amigo.ai. For Platform API equivalents, start with the Platform SDK Quickstart.

1. Conversation Lifecycle Management

Create a conversation, send messages, handle streaming events, and finish cleanly.

Python

from amigo_sdk import AsyncAmigoClient
from amigo_sdk.models import (
    ConversationCreateConversationRequest,
    CreateConversationParametersQuery,
    InteractWithConversationParametersQuery,
    Format,
)

async def run_conversation():
    async with AsyncAmigoClient() as client:
        # 1. Create conversation (service_id goes in the request body)
        params = CreateConversationParametersQuery(
            response_format="text",
        )
        body = ConversationCreateConversationRequest(service_id="your-service-id")

        conversation_id = None
        async for event in await client.conversations.create_conversation(body, params):
            if hasattr(event, "conversation_id"):
                conversation_id = event.conversation_id
            # Handle greeting events...

        # 2. Interact
        interact_params = InteractWithConversationParametersQuery(
            request_format=Format.text,
            response_format="text",
        )
        async for event in await client.conversations.interact_with_conversation(
            conversation_id,
            interact_params,
            text_message="Hello, I need help with my account",
        ):
            print(event)

        # 3. Finish
        await client.conversations.finish_conversation(conversation_id)

TypeScript

Key considerations:

• Always call finish_conversation when done to release server resources.
• Handle the NDJSON stream event by event. Don't buffer the entire response.
• Use abort_event (Python) or AbortController (TypeScript) for cancellation.

---

2. User Creation and User Model Queries

Create users, let conversations enrich the user model, and query the result. To push facts from your own systems into the model, use additional_context - see Enriching the User Model.

Python

from amigo_sdk import AmigoClient
from amigo_sdk.models import UserCreateInvitedUserRequest

client = AmigoClient()

# Create user. role_name must be a role returned by the Get roles endpoint.
# Optional fields: login_link, nonsensitive_user_variables,
# sensitive_user_variables, user_preferences.
user = client.users.create_user(
    UserCreateInvitedUserRequest(
        first_name="Jane",
        last_name="Doe",
        email="[e-mail]",
        role_name="DefaultUserRole",
    )
)
print(f"Created user: {user.user_id}")

# After conversations have occurred, query the enriched model
model = client.users.get_user_model(user.user_id)
print(f"User model dimensions: {model}")

TypeScript

---

3. Webhook-Driven Memory Sync

Receive conversation events via webhooks and sync memories to an external system.

Python (FastAPI)

import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
WEBHOOK_SECRET = "your-webhook-secret"

@app.post("/webhook")
async def handle_webhook(request: Request):
    body = await request.body()
    timestamp = request.headers.get("x-amigo-request-timestamp", "")
    signature = request.headers.get("x-amigo-request-signature", "")

    # Verify signature (constant-time comparison)
    payload = f"v1:{timestamp}:{body.decode()}"
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), payload.encode(), hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail="Invalid signature")

    data = await request.json()
    event_type = data.get("type")

    if event_type == "conversation-post-processing-complete":
        # Sync updated memories to external system
        await sync_to_external_db(data)

    return {"status": "ok"}

TypeScript (Express)

Key considerations:

• Always use constant-time comparison for signature verification.
• Respond with 200 quickly and process events asynchronously if needed.
• Implement idempotency using the x-amigo-idempotent-key header, which stays constant across retries.
• Event payload shapes are documented in Webhook Event Types.

---

4. Multi-Service Routing

List available services and route conversations to the appropriate one based on user needs.

Python

from amigo_sdk import AmigoClient
from amigo_sdk.models import GetServicesParametersQuery

client = AmigoClient()

# List all services
services = client.services.get_services(GetServicesParametersQuery())

# Route based on user intent
def select_service(user_intent: str, services):
    for svc in services.services:
        if user_intent.lower() in svc.description.lower():
            return svc.id
    return services.services[0].id  # fallback to first

service_id = select_service("billing question", services)
# Now create conversation with the selected service_id

TypeScript

---

5. Simulation-Driven Deployment

Run automated simulations before deploying agent changes to production. Unlike the patterns above, this one uses the Agent Forge CLI (forge) rather than the SDKs - the CLI wraps the simulation APIs and is the recommended way to gate deployments in CI.

Python

import subprocess
import sys

def run_simulation_gate():
    """CI gate that runs simulations before deploy."""
    # Use Agent Forge CLI for simulation
    result = subprocess.run(
        [
            "forge", "conversation", "simulate",
            "-s", "My Service",
            "-e", "staging",
            "--test-user",
            "--analyze-wait", "30",
            "--max-turns", "10",
        ],
        capture_output=True, text=True,
    )

    if result.returncode != 0:
        print(f"Simulation failed:\n{result.stderr}")
        sys.exit(1)

    print("Simulation passed. Safe to deploy.")

if __name__ == "__main__":
    run_simulation_gate()

GitHub Actions

Key considerations:

• Use --test-user for parallel simulation execution without conflicts.
• Set --analyze-wait to allow time for metrics computation after simulation.
• Define clear pass/fail thresholds for your metrics (for example, safety >= 9.0).