Skip to main content
POST
https://sandbox.getcollate.io/api
/
{agent_name}
/
invoke
POST /{agent_name}/invoke
from ai_sdk import AISdk, AISdkConfig

config = AISdkConfig.from_env()
client = AISdk.from_config(config)

# Single invocation
response = client.agent("DataQualityPlannerAgent").call(
    "What data quality tests should I add for the customers table?"
)
print(response.response)
print(f"Conversation ID: {response.conversation_id}")
print(f"Tools used: {response.tools_used}")

{
  "conversationId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "response": "Based on my analysis of the `customers` table, I recommend the following data quality tests:\n\n1. **Not Null** on `customer_id` — primary key should never be null\n2. **Unique** on `email` — email addresses must be unique\n3. **Column Values to Be Between** on `age` — expected range 0-150\n4. **Column Values to Match Regex** on `email` — validate email format",
  "toolsUsed": [
    "search_metadata",
    "get_entity_details"
  ],
  "usage": {
    "promptTokens": 1250,
    "completionTokens": 340,
    "totalTokens": 1590
  }
}

​
Invoke Agent

Send a message to an AI Studio Agent and receive a response. The AI SDK supports single-turn invocations, streaming responses, and multi-turn conversations with automatic context management.

​
Path Parameters

​
agent_name
string
required
Name of the AI Studio Agent to invoke. Must match an API-enabled agent in your Collate instance. Case-sensitive.

​
Body Parameters

​
message
string
required
The message or prompt to send to the agent.
​
conversationId
string
Unique conversation identifier for multi-turn follow-ups. Omit for a new conversation. The agent returns a conversationId in every response that you can pass in subsequent requests to maintain context.
​
parameters
object
Optional key-value parameters to pass to the agent for customizing behavior.
POST /{agent_name}/invoke
from ai_sdk import AISdk, AISdkConfig

config = AISdkConfig.from_env()
client = AISdk.from_config(config)

# Single invocation
response = client.agent("DataQualityPlannerAgent").call(
    "What data quality tests should I add for the customers table?"
)
print(response.response)
print(f"Conversation ID: {response.conversation_id}")
print(f"Tools used: {response.tools_used}")

{
  "conversationId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "response": "Based on my analysis of the `customers` table, I recommend the following data quality tests:\n\n1. **Not Null** on `customer_id` — primary key should never be null\n2. **Unique** on `email` — email addresses must be unique\n3. **Column Values to Be Between** on `age` — expected range 0-150\n4. **Column Values to Match Regex** on `email` — validate email format",
  "toolsUsed": [
    "search_metadata",
    "get_entity_details"
  ],
  "usage": {
    "promptTokens": 1250,
    "completionTokens": 340,
    "totalTokens": 1590
  }
}

​
Response

​
conversationId
string
Unique conversation identifier for multi-turn follow-ups. Pass this value in subsequent requests to maintain context.
​
response
string
The agent’s text response to your message.
​
toolsUsed
array
List of tool names the agent invoked while processing the request.
​
usage
object
Token usage information for the request.

​
Streaming

Use the streaming endpoint to receive real-time output as the agent generates its response. This is useful for long-running requests or when you want to display incremental results in a UI. The streaming endpoint is POST /{agent_name}/stream and uses Server-Sent Events (SSE).
POST /{agent_name}/stream
from ai_sdk import AISdk, AISdkConfig

config = AISdkConfig.from_env()
client = AISdk.from_config(config)

for event in client.agent("DataQualityPlannerAgent").stream(
    "Analyze the orders table"
):
    match event.type:
        case "start":
            print(f"Started conversation: {event.conversation_id}")
        case "content":
            print(event.content, end="", flush=True)
        case "tool_use":
            print(f"\n[Using tool: {event.tool_name}]")
        case "end":
            print("\nDone!")
        case "error":
            print(f"\nError: {event.error}")

​
Stream Event Types

TypeFieldsDescription
startconversation_idAgent started processing the request
contentcontentText chunk from the agent’s response
tool_usetool_nameAgent is invoking an MCP tool
end-Agent finished generating the response
errorerrorAn error occurred during processing

​
Multi-Turn Conversations

The Conversation class automatically manages conversation context across messages, passing the conversationId between turns so the agent retains full history. 
from ai_sdk import AISdk, AISdkConfig, Conversation

config = AISdkConfig.from_env()
client = AISdk.from_config(config)

conv = Conversation(client.agent("DataQualityPlannerAgent"))

# Each call automatically carries the conversation context
print(conv.send("Analyze the customers table"))
print(conv.send("Create tests for the issues you found"))
print(conv.send("Show me the SQL for those tests"))

# Access conversation details
print(f"Turns: {len(conv)}")
print(f"Conversation ID: {conv.id}")
In TypeScript: 
const conv = client.agent('DataQualityPlannerAgent').conversation();

const r1 = await conv.send('Analyze the customers table');
console.log(r1.response);

const r2 = await conv.send('Create tests for the issues you found');
console.log(r2.response);

console.log(`Turns: ${conv.turns}`);
console.log(`Conversation ID: ${conv.id}`);

​
Async Support

All sync methods have async counterparts with the a prefix (Python) or are natively async (TypeScript):
SyncAsync
agent.call()await agent.acall()
agent.stream()await agent.astream()
conv.send()await conv.asend()
import asyncio
from ai_sdk import AISdk, AISdkConfig

async def main():
    config = AISdkConfig.from_env(enable_async=True)
    client = AISdk.from_config(config)

    try:
        response = await client.agent("DataQualityPlannerAgent").acall(
            "Analyze the customers table"
        )
        print(response.response)
    finally:
        await client.aclose()
        client.close()

asyncio.run(main())

​
Error Handling

CodeExceptionDescription
401AuthenticationErrorInvalid or expired JWT token
403AgentNotEnabledErrorAgent exists but is not API-enabled
404AgentNotFoundErrorNo agent with the given name exists
429RateLimitErrorToo many requests — retry after the indicated delay
500AgentExecutionErrorInternal error during agent execution
from ai_sdk.exceptions import (
    AuthenticationError,
    AgentNotFoundError,
    AgentNotEnabledError,
    RateLimitError,
    AgentExecutionError,
)

try:
    response = client.agent("MyAgent").call("Hello")

except AuthenticationError:
    print("Invalid or expired token. Check your AI_SDK_TOKEN.")

except AgentNotFoundError as e:
    print(f"Agent not found: {e.agent_name}")

except AgentNotEnabledError as e:
    print(f"Agent '{e.agent_name}' is not API-enabled. Enable it in AI Studio.")

except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after} seconds")

except AgentExecutionError as e:
    print(f"Agent execution failed: {e.message}")