Common Patterns

Production-ready patterns for conversation lifecycle, user enrichment, webhooks, and routing.

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

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
        params = CreateConversationParametersQuery(
            service_id="your-service-id",
            response_format="text",
        )
        body = ConversationCreateConversationRequest()

        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 Model Enrichment Pipeline

Create users, attach context, and query their enriched user model.

Python

from amigo_sdk import AmigoClient
from amigo_sdk.models import UserCreateInvitedUserRequest

client = AmigoClient()

# Create user with initial context
user = client.users.create_user(
    UserCreateInvitedUserRequest(
        user_name="jane.doe",
        display_name="Jane Doe",
    )
)
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("event_type")

    if event_type == "conversation.interaction.completed":
        # Sync interaction data 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 -- process events asynchronously if needed

• Implement idempotency using the event ID to handle retries

---

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.results:
        if user_intent.lower() in svc.description.lower():
            return svc.service_id
    return services.results[0].service_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.

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 (e.g., safety >= 9.0)