# Common Patterns

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.

{% @mermaid/diagram content="%%{init: {"theme": "base", "themeVariables": {"actorBkg": "#083241", "actorTextColor": "#FFFFFF", "actorBorder": "#083241", "signalColor": "#575452", "signalTextColor": "#100F0F", "labelBoxBkgColor": "#F1EAE7", "labelBoxBorderColor": "#D7D2D0", "labelTextColor": "#100F0F", "loopTextColor": "#100F0F", "noteBkgColor": "#F1EAE7", "noteBorderColor": "#D7D2D0", "noteTextColor": "#100F0F", "activationBkgColor": "#E8E2EB", "activationBorderColor": "#083241", "altSectionBkgColor": "#F1EAE7", "altSectionColor": "#100F0F"}}}%%
sequenceDiagram
participant App
participant Amigo API
App->>Amigo API: POST /conversation/ (NDJSON stream)
Amigo API-->>App: conversation-created event
Amigo API-->>App: agent greeting events
App->>Amigo API: POST /conversation/{id}/interact (text)
Amigo API-->>App: agent response events
Amigo API-->>App: interaction-complete event
App->>Amigo API: POST /conversation/{id}/finish/" %}

{% tabs %}
{% tab title="Python" %}

```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)
```

{% endtab %}

{% tab title="TypeScript" %}

```typescript
import { AmigoClient, conversationId } from '@amigo-ai/sdk'

const client = new AmigoClient({
  apiKey: process.env.AMIGO_API_KEY!,
  apiKeyId: process.env.AMIGO_API_KEY_ID!,
  userId: process.env.AMIGO_USER_ID!,
  orgId: process.env.AMIGO_ORG_ID!,
})

// 1. Create conversation
let convId: string | undefined
for await (const event of await client.conversations.createConversation(
  {},
  { service_id: 'your-service-id', response_format: 'text' }
)) {
  if ('conversation_id' in event) convId = event.conversation_id
}

// 2. Interact
for await (const event of await client.conversations.interactWithConversation(
  conversationId(convId!),
  'Hello, I need help with my account',
  { request_format: 'text', response_format: 'text' }
)) {
  console.log(event)
}

// 3. Finish
await client.conversations.finishConversation(conversationId(convId!))
```

{% endtab %}
{% endtabs %}

**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.

{% @mermaid/diagram content="%%{init: {"theme": "base", "themeVariables": {"actorBkg": "#083241", "actorTextColor": "#FFFFFF", "actorBorder": "#083241", "signalColor": "#575452", "signalTextColor": "#100F0F", "labelBoxBkgColor": "#F1EAE7", "labelBoxBorderColor": "#D7D2D0", "labelTextColor": "#100F0F", "loopTextColor": "#100F0F", "noteBkgColor": "#F1EAE7", "noteBorderColor": "#D7D2D0", "noteTextColor": "#100F0F", "activationBkgColor": "#E8E2EB", "activationBorderColor": "#083241", "altSectionBkgColor": "#F1EAE7", "altSectionColor": "#100F0F"}}}%%
sequenceDiagram
participant App
participant Amigo API
App->>Amigo API: POST /user/ (create user)
Amigo API-->>App: User created
App->>Amigo API: POST /conversation/ (with user)
Note over Amigo API: Conversations enrich user model
App->>Amigo API: GET /user/{id}/model/
Amigo API-->>App: Enriched user model" %}

{% tabs %}
{% tab title="Python" %}

```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}")
```

{% endtab %}

{% tab title="TypeScript" %}

```typescript
// Create user
const user = await client.users.create({ user_name: 'jane.doe', display_name: 'Jane Doe' })
console.log('Created:', user.user_id)

// After conversations, query enriched model
const model = await client.users.getUserModel(userId(user.user_id))
console.log('User model:', model)
```

{% endtab %}
{% endtabs %}

***

## 3. Webhook-Driven Memory Sync

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

{% @mermaid/diagram content="%%{init: {"theme": "base", "themeVariables": {"actorBkg": "#083241", "actorTextColor": "#FFFFFF", "actorBorder": "#083241", "signalColor": "#575452", "signalTextColor": "#100F0F", "labelBoxBkgColor": "#F1EAE7", "labelBoxBorderColor": "#D7D2D0", "labelTextColor": "#100F0F", "loopTextColor": "#100F0F", "noteBkgColor": "#F1EAE7", "noteBorderColor": "#D7D2D0", "noteTextColor": "#100F0F", "activationBkgColor": "#E8E2EB", "activationBorderColor": "#083241", "altSectionBkgColor": "#F1EAE7", "altSectionColor": "#100F0F"}}}%%
sequenceDiagram
participant Amigo API
participant Webhook Endpoint
participant External DB
Amigo API->>Webhook Endpoint: POST /webhook (signed)
Webhook Endpoint->>Webhook Endpoint: Verify HMAC signature
Webhook Endpoint->>External DB: Store event data
Webhook Endpoint-->>Amigo API: 200 OK" %}

{% tabs %}
{% tab title="Python (FastAPI)" %}

```python
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"}
```

{% endtab %}

{% tab title="TypeScript (Express)" %}

```typescript
import crypto from 'crypto'
import express from 'express'

const app = express()
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET!

app.post('/webhook', express.raw({ type: '*/*' }), (req, res) => {
  const timestamp = req.headers['x-amigo-request-timestamp'] as string
  const signature = req.headers['x-amigo-request-signature'] as string
  const body = req.body.toString()

  // Verify signature (constant-time comparison)
  const payload = `v1:${timestamp}:${body}`
  const expected = crypto.createHmac('sha256', WEBHOOK_SECRET).update(payload).digest('hex')

  if (!crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))) {
    return res.status(401).send('Invalid signature')
  }

  const data = JSON.parse(body)
  // Process event...
  res.json({ status: 'ok' })
})
```

{% endtab %}
{% endtabs %}

**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.

{% @mermaid/diagram content="%%{init: {"theme": "base", "themeVariables": {"actorBkg": "#083241", "actorTextColor": "#FFFFFF", "actorBorder": "#083241", "signalColor": "#575452", "signalTextColor": "#100F0F", "labelBoxBkgColor": "#F1EAE7", "labelBoxBorderColor": "#D7D2D0", "labelTextColor": "#100F0F", "loopTextColor": "#100F0F", "noteBkgColor": "#F1EAE7", "noteBorderColor": "#D7D2D0", "noteTextColor": "#100F0F", "activationBkgColor": "#E8E2EB", "activationBorderColor": "#083241", "altSectionBkgColor": "#F1EAE7", "altSectionColor": "#100F0F"}}}%%
sequenceDiagram
participant User
participant Router
participant Amigo API
User->>Router: Request help
Router->>Amigo API: GET /service/ (list services)
Amigo API-->>Router: Available services
Router->>Router: Select best service
Router->>Amigo API: POST /conversation/ (with selected service)
Amigo API-->>Router: Conversation stream
Router-->>User: Agent response" %}

{% tabs %}
{% tab title="Python" %}

```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
```

{% endtab %}

{% tab title="TypeScript" %}

```typescript
const services = await client.services.list()

function selectService(intent: string) {
  const match = services.results?.find(s =>
    s.description?.toLowerCase().includes(intent.toLowerCase())
  )
  return match?.service_id ?? services.results?.[0]?.service_id
}

const serviceId = selectService('billing question')
```

{% endtab %}
{% endtabs %}

***

## 5. Simulation-Driven Deployment

Run automated simulations before deploying agent changes to production.

{% @mermaid/diagram content="%%{init: {"theme": "base", "themeVariables": {"actorBkg": "#083241", "actorTextColor": "#FFFFFF", "actorBorder": "#083241", "signalColor": "#575452", "signalTextColor": "#100F0F", "labelBoxBkgColor": "#F1EAE7", "labelBoxBorderColor": "#D7D2D0", "labelTextColor": "#100F0F", "loopTextColor": "#100F0F", "noteBkgColor": "#F1EAE7", "noteBorderColor": "#D7D2D0", "noteTextColor": "#100F0F", "activationBkgColor": "#E8E2EB", "activationBorderColor": "#083241", "altSectionBkgColor": "#F1EAE7", "altSectionColor": "#100F0F"}}}%%
sequenceDiagram
participant CI
participant Amigo API
participant Metrics
CI->>Amigo API: Create test personas & scenarios
CI->>Amigo API: Run simulation test set
Amigo API-->>CI: Simulation results
CI->>Metrics: Evaluate pass/fail criteria
alt Metrics pass
CI->>CI: Deploy to production
else Metrics fail
CI->>CI: Block deployment
end" %}

{% tabs %}
{% tab title="Python" %}

```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()
```

{% endtab %}

{% tab title="GitHub Actions" %}

```yaml
jobs:
  simulation-gate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Install Agent Forge
        run: pip install agent-forge
      - name: Run simulations
        env:
          API_KEY: ${{ secrets.AMIGO_API_KEY }}
          API_KEY_ID: ${{ secrets.AMIGO_API_KEY_ID }}
        run: |
          forge conversation simulate \
            -s "${{ vars.SERVICE_NAME }}" \
            -e staging \
            --test-user \
            --max-turns 10
      - name: Deploy if passed
        if: success()
        run: echo "Deploying..."
```

{% endtab %}
{% endtabs %}

**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)