Skip to main content

​
Agent Integration

Your agent backend now receives requests from Convoy instead of directly from your client. You need to do two things:
  1. Verify the request — check Convoy’s HMAC signature
  2. Report metrics — send latency, outcome, cost, and input/output to Convoy

​
1. Verify the request

Convoy signs every request it forwards using HMAC-SHA256 with your shared secret. Your agent must verify this signature to confirm the request came from Convoy.
If signature verification fails, reject the request immediately — return a 401 and do not run your agent. If you also need to accept direct callers on this route, add your own auth (e.g. API key or session token).

​
Headers Convoy adds

HeaderDescription
X-Convoy-SignatureHMAC signature: t={timestamp},v1={hmac}
X-Convoy-Agent-NameYour agent’s name
X-Convoy-Version-NameWhich version this request was routed to (e.g. v1, v2-test)
Session-IDThe session ID from the client

​
How to verify

  1. Parse X-Convoy-Signature to extract t (timestamp), v1 (HMAC), and optionally v2 (HMAC)
  2. Reject if timestamp is older than 5 minutes
  3. Build the signed payload: {timestamp}\n{METHOD}\n{path_with_query}\n{body} For example, a POST to /api/chat?session=abc with body {"msg":"hi"} at timestamp 1700000000: 
    1700000000\nPOST\n/api/chat?session=abc\n{"msg":"hi"}
    {path_with_query} is the path of the request as it arrives at your backend, including any query string. Convoy constructs this by concatenating the version’s base path with the client’s request path — for example, a version URL of /api/chat plus a client path of /completions produces /api/chat/completions.
  4. Compute HMAC-SHA256 using your secret
  5. Accept if either v1 or v2 matches
import hmac
import hashlib
import time

def verify_convoy_signature(signature_header: str, secret: str,
                            method: str, path: str, body: str):
    """Verify HMAC signature from Convoy proxy."""
    parts = dict(p.split("=", 1) for p in signature_header.split(","))
    timestamp = int(parts["t"])

    if abs(time.time() - timestamp) > 300:
        raise ValueError("Signature expired")

    payload = f"{timestamp}\n{method.upper()}\n{path}\n{body}"
    expected = hmac.new(
        secret.strip().encode(), payload.encode(), hashlib.sha256
    ).hexdigest()

    # v1 = current secret, v2 = previous secret (during rotation)
    # Your env has one secret — it will match whichever version was signed with it
    for v in ["v1", "v2"]:
        if v in parts and hmac.compare_digest(parts[v], expected):
            return
    raise ValueError("Invalid signature")
During secret rotation, Convoy sends both v1 (signed with the new secret) and v2 (signed with the old secret) for 24 hours. Your env has one secret — it will match one of them regardless of whether you’ve rotated yet. The code above handles this automatically.

​
2. Report metrics

After your agent processes the request, report metrics to Convoy. This data powers test evaluation — Convoy uses it to decide whether to promote or roll back. Endpoint: POST https://api.convoylabs.com/v1/ingest/sessions Auth: Authorization: Bearer <CONVOY_SECRET> (same secret)

​
Payload

Always required:
FieldTypeDescription
session_idstringFrom the Session-ID header
agent_namestringFrom the X-Convoy-Agent-Name header
version_namestringFrom the X-Convoy-Version-Name header
outcome"success" | "error"Did this step succeed or fail?
is_last_stepbooleanIs this the final step of the session?
step_indexintegerZero-based step index in the session (default: 0)
attempt_numberintegerAttempt number for this step, 1-based (default: 1)
Required on success, optional on error:
FieldTypeDescription
latency_msintegerHow long the agent took (milliseconds)
cost_usdfloatToken/API cost in USD (0 if none)
Required on last step (unless outcome is "error"):
FieldTypeDescription
inputstringThe input to the agent — allowed on any step, required on the last successful step
outputstringThe agent’s response — only allowed when is_last_step is true
When outcome is "error", the call may have crashed before producing cost, latency, input, or output. Include these fields if available, but omit them if the data doesn’t exist — don’t fabricate values.

​
Rules

  • input and output are required on the last step unless outcome is "error"
  • output is forbidden when is_last_step is false
  • input is allowed on any step — it overwrites the session’s stored input
  • On success, output should contain everything you want the judge prompt to evaluate — typically the agent’s response, and optionally reasoning or tool calls
  • For multi-step sessions (e.g. resumed or suspendable workflows), increment step_index per step (0, 1, 2…) and report the final step with is_last_step: true
  • If your agent retries a step, report each attempt with the same step_index but increment attempt_number (1, 2, 3…). All attempts are recorded

​
Code examples

import asyncio
import httpx
import time
import os

async def handle_agent_request(request):
    # 1. Verify signature — reject immediately if invalid
    try:
        verify_convoy_signature(
            request.headers["X-Convoy-Signature"],
            os.environ["CONVOY_SECRET"].strip(),
            request.method,
            request.url.path + ("?" + request.url.query if request.url.query else ""),
            await request.body(),
        )
    except (ValueError, KeyError):
        return JSONResponse(status_code=401, content={"error": "Unauthorized"})

    # 2. Extract Convoy headers
    session_id = request.headers["Session-ID"]
    agent_name = request.headers["X-Convoy-Agent-Name"]
    version_name = request.headers["X-Convoy-Version-Name"]

    # 3. Run your agent and measure
    start = time.time()
    result = await run_agent(request)  # your agent logic
    latency_ms = int((time.time() - start) * 1000)

    # 4. Report metrics (fire-and-forget with retries)
    payload = {
        "session_id": session_id,
        "agent_name": agent_name,
        "version_name": version_name,
        "outcome": "success" if result.ok else "error",
        "is_last_step": result.is_done,
        "step_index": result.step_index,
        "attempt_number": result.attempt_number,
    }
    if result.ok:
        payload["input"] = result.input
        payload["latency_ms"] = latency_ms
        payload["cost_usd"] = result.token_cost
        if result.is_done:
            payload["output"] = result.response
    else:
        # On error, include what's available — don't fabricate values
        if result.input:
            payload["input"] = result.input
        if latency_ms:
            payload["latency_ms"] = latency_ms

    async def report():
        async with httpx.AsyncClient() as client:
            for attempt in range(3):
                try:
                    r = await client.post(
                        "https://api.convoylabs.com/v1/ingest/sessions",
                        headers={"Authorization": f"Bearer {os.environ['CONVOY_SECRET'].strip()}"},
                        json=payload,
                    )
                    if r.status_code < 500:
                        return
                except httpx.TransportError:
                    pass
                await asyncio.sleep(0.5 * 2 ** attempt)

    asyncio.create_task(report())

    # 5. Return response to user
    return result.response
Report metrics asynchronously so it doesn’t block your agent’s response. The examples above retry up to 3 times with exponential backoff. Duplicate (session_id, step_index, attempt_number) reports safely overwrite previous data.
Serverless runtimes (Vercel, AWS Lambda, Cloud Functions) may terminate your function immediately after you return the response, killing any in-flight fire-and-forget requests. To avoid losing metrics, either await the ingest call (with a short timeout) before returning, or use your platform’s execution extension (e.g. Vercel’s waitUntil).