Skip to content

Direct Agent Endpoint

The direct endpoint lets you send messages to a specific agent by ID, bypassing load balancing.

Endpoint

POST /api/v1/agents/{agent_id}/messages

When to Use

Use the direct endpoint when:

  • Testing a specific agent
  • Debugging agent behavior
  • Targeting a specialized agent
  • Building custom routing logic

For production with multiple agents, prefer the Pool Endpoint.

Request

Headers

Header Required Description
X-API-Key Yes Your API key
Content-Type Yes application/json

Path Parameters

Parameter Type Description
agent_id UUID The specific agent's ID

Body (Synchronous)

{
  "message": "Your message here"
}

Body (Asynchronous with Webhook)

{
  "message": "Your message here",
  "webhook_url": "https://your-server.com/webhook",
  "session_id": "optional-session-id"
}
Field Type Required Description
message string Yes Message content (max 10,000 chars)
webhook_url string No URL for async response delivery
session_id string No Session ID for conversation tracking

Response

Synchronous (200 OK)

{
  "message_id": "msg_abc123",
  "content": [
    {
      "type": "text",
      "text": "Hello! How can I help you today?"
    }
  ],
  "metadata": {
    "tokens_used": 45,
    "model": "gpt-4o",
    "duration_ms": 1250,
    "tool_calls": []
  }
}

Asynchronous (202 Accepted)

{
  "callback_id": "cb_xyz789",
  "status": "processing",
  "agent_id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2025-01-15T10:30:00Z"
}

Content Blocks

Responses contain an array of content blocks:

Text Block

{
  "type": "text",
  "text": "The response text content"
}

Error Block

{
  "type": "error",
  "error": "An error occurred",
  "details": "Additional error information"
}

Examples

Synchronous Request

curl -X POST https://agents.ag2trust.com/api/v1/agents/{agent_id}/messages \
  -H "X-API-Key: cust_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello, what can you do?"}'

import requests

agent_id = "550e8400-e29b-41d4-a716-446655440000"

response = requests.post(
    f"https://agents.ag2trust.com/api/v1/agents/{agent_id}/messages",
    headers={
        "X-API-Key": "cust_your_api_key",
        "Content-Type": "application/json"
    },
    json={"message": "Hello, what can you do?"},
    timeout=65
)

data = response.json()
for block in data["content"]:
    if block["type"] == "text":
        print(block["text"])

const agentId = '550e8400-e29b-41d4-a716-446655440000';

const response = await fetch(
  `https://agents.ag2trust.com/api/v1/agents/${agentId}/messages`,
  {
    method: 'POST',
    headers: {
      'X-API-Key': 'cust_your_api_key',
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      message: 'Hello, what can you do?'
    })
  }
);

const data = await response.json();
data.content.forEach(block => {
  if (block.type === 'text') {
    console.log(block.text);
  }
});

Asynchronous Request

curl -X POST https://agents.ag2trust.com/api/v1/agents/{agent_id}/messages \
  -H "X-API-Key: cust_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "message": "Analyze this large dataset",
    "webhook_url": "https://your-server.com/webhook"
  }'

response = requests.post(
    f"https://agents.ag2trust.com/api/v1/agents/{agent_id}/messages",
    headers={
        "X-API-Key": "cust_your_api_key",
        "Content-Type": "application/json"
    },
    json={
        "message": "Analyze this large dataset",
        "webhook_url": "https://your-server.com/webhook"
    }
)

data = response.json()
print(f"Processing started: {data['callback_id']}")
# Response will be delivered to webhook_url

Finding Agent IDs

Via API

curl https://agents.ag2trust.com/api/v1/agents \
  -H "X-API-Key: cust_your_api_key"

Response:

{
  "agents": [
    {
      "agent_id": "550e8400-e29b-41d4-a716-446655440000",
      "name": "support-bot-1",
      "status": "running",
      "capabilities": ["web_http"]
    }
  ]
}

Via Dashboard

  1. Go to Agents page
  2. Click on an agent
  3. Find the ID in the URL or agent details

Error Responses

401 Unauthorized

{
  "error": "Invalid or missing API key",
  "error_code": "INVALID_API_KEY"
}

403 Forbidden

{
  "error": "Agent not found or not owned by customer",
  "error_code": "AGENT_NOT_FOUND"
}

The agent doesn't exist or belongs to a different organization.

404 Not Found

{
  "error": "Agent does not exist",
  "error_code": "AGENT_NOT_FOUND"
}

400 Bad Request - Agent Not Running

{
  "error": "Agent is not running",
  "error_code": "AGENT_NOT_RUNNING"
}

Start the agent before sending messages.

500 Internal Server Error

{
  "error": "Agent error occurred",
  "error_code": "INTERNAL_ERROR",
  "details": "Error details..."
}

Timeout Handling

Synchronous requests timeout after 60 seconds:

try:
    response = requests.post(
        url,
        json=payload,
        headers=headers,
        timeout=65  # Client timeout slightly > server timeout
    )
except requests.Timeout:
    print("Request timed out - consider using async with webhook")

For long-running tasks, use async mode with webhooks.

Response Metadata

The metadata object provides useful information:

Field Type Description
tokens_used integer Total tokens consumed
model string Model used (e.g., "gpt-4o")
duration_ms integer Processing time in milliseconds
tool_calls array Tools invoked during processing

Tool Calls Example

{
  "metadata": {
    "tool_calls": [
      {
        "tool": "web_search",
        "arguments": {"query": "weather today"},
        "result": "..."
      }
    ]
  }
}

Best Practices

1. Verify Agent Status First

# Check agent is running before sending messages
agents_response = requests.get(
    f"https://agents.ag2trust.com/api/v1/agents",
    headers={"X-API-Key": api_key}
)

agent = next(
    (a for a in agents_response.json()["agents"]
     if a["agent_id"] == agent_id),
    None
)

if not agent or agent["status"] != "running":
    raise Exception("Agent is not available")

2. Use Appropriate Timeouts

Task Type Recommended Timeout
Simple Q&A 30s
With tools 60s
Complex analysis Use async/webhook

3. Handle All Content Block Types

for block in response["content"]:
    if block["type"] == "text":
        handle_text(block["text"])
    elif block["type"] == "error":
        handle_error(block["error"], block.get("details"))

Comparison with Pool Endpoint

Feature Direct Endpoint Pool Endpoint
Target specific agent Yes No
Load balancing No Yes
Built-in threading No Yes
Async/webhook Yes No (sync only)
Best for Testing, specific routing Production traffic

Next Steps