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 should verify this signature to confirm the request came from Convoy.

​
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}
  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.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

FieldTypeRequiredDescription
session_idstringalwaysFrom the Session-ID header
agent_namestringalwaysFrom the X-Convoy-Agent-Name header
version_namestringalwaysFrom the X-Convoy-Version-Name header
outcome"success" | "error"alwaysDid this step succeed or fail?
is_last_stepbooleanalwaysIs this the final step of the session?
latency_msintegeralwaysHow long the agent took (milliseconds)
cost_usdfloatalwaysToken/API cost in USD (0 if none)
inputstringwhen is_last_stepThe input to the agent
outputstringwhen is_last_stepThe agent’s response — everything you want the judge to evaluate
step_indexintegeroptionalZero-based step index in the session (default: 0)
attempt_numberintegeroptionalAttempt number for this step, 1-based (default: 1)

​
Rules

  • output is only allowed when is_last_step is true
  • When is_last_step is true, both input and output are required
  • 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 (conversational agents, 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 httpx
import time
import os

async def handle_agent_request(request):
    # 1. Verify signature (see above)
    verify_convoy_signature(
        request.headers["X-Convoy-Signature"],
        os.environ["CONVOY_SECRET"],
        request.method,
        request.url.path,
        await request.body(),
    )

    # 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 to not block the response)
    payload = {
        "session_id": session_id,
        "agent_name": agent_name,
        "version_name": version_name,
        "input": result.input,
        "outcome": "success" if result.ok else "error",
        "is_last_step": result.is_done,
        "latency_ms": latency_ms,
        "cost_usd": result.token_cost,
        "step_index": result.step_index,        # optional, default 0
        "attempt_number": result.attempt_number,  # optional, default 1
    }
    if result.is_done:
        payload["output"] = result.response

    async with httpx.AsyncClient() as client:
        await client.post(
            "https://api.convoylabs.com/v1/ingest/sessions",
            headers={"Authorization": f"Bearer {os.environ['CONVOY_SECRET']}"},
            json=payload,
        )

    # 5. Return response to user
    return result.response
Report metrics asynchronously (fire-and-forget) so it doesn’t block your agent’s response to the user. Reporting the same (session_id, step_index, attempt_number) twice overwrites the previous data — safe to retry on network errors. For production, add retries with exponential backoff on 5xx and timeout errors to avoid losing metric data.