Skip to main content
POST
https://sandbox.open-metadata.org/api
/
{agent_name}
/
invoke
POST /{agent_name}/invoke
from ai_sdk import AISdk, AISdkConfig

config = AISdkConfig.from_env()  # reads AI_SDK_HOST and AI_SDK_TOKEN
client = AISdk.from_config(config)

# Invoke an agent
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}")

{
  "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\n5. **Table Row Count to Be Between** — expected range 1000-5000000",
  "conversation_id": "conv-a1b2c3d4-e5f6-7890-abcd-ef1234567890"
}

​
Invoke Agent

Send a message to an AI Studio Agent and receive a response. Supports both synchronous and streaming modes, as well as multi-turn conversations using a conversationId.

​
Path Parameters

​
agent_name
string
required
Name of the AI Studio Agent to invoke (case-sensitive). Must be an API-enabled agent.

​
Body Parameters

​
message
string
required
The message to send to the agent.
​
conversationId
string
ID of an existing conversation to continue. Omit to start a new conversation. The agent retains full context from previous turns.
​
parameters
object
Optional key-value parameters to pass to the agent for customization.
POST /{agent_name}/invoke
from ai_sdk import AISdk, AISdkConfig

config = AISdkConfig.from_env()  # reads AI_SDK_HOST and AI_SDK_TOKEN
client = AISdk.from_config(config)

# Invoke an agent
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}")

{
  "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\n5. **Table Row Count to Be Between** — expected range 1000-5000000",
  "conversation_id": "conv-a1b2c3d4-e5f6-7890-abcd-ef1234567890"
}

​
Returns

Returns an InvokeResponse containing the agent’s text response and the conversation ID for follow-up messages.

​
Response

​
response
string
The agent’s text response to the message.
​
conversation_id
string
Unique identifier for the conversation. Use this in subsequent requests to continue the same conversation thread.

​
Streaming

Stream responses in real time as the agent generates them. Instead of waiting for the full response, you receive a series of server-sent events (SSE).
POST /{agent_name}/invoke (streaming)
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
contentcontentText chunk from the agent
tool_usetool_nameAgent is using a tool
end-Agent finished
errorerrorError occurred

​
Multi-Turn Conversations

Continue an existing conversation by passing the conversationId from a previous response. The agent retains full context from all prior turns in the conversation. 
from ai_sdk import Conversation

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

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}")
Alternatively, manage the conversation ID manually: 
# First turn
response = client.agent("DataQualityPlannerAgent").call(
    "Analyze the customers table"
)
conv_id = response.conversation_id

# Follow-up turns reuse the conversation ID
response = client.agent("DataQualityPlannerAgent").call(
    "Create tests for the issues you found",
    conversation_id=conv_id
)

​
Error Handling

CodeError TypeDescription
400BAD_REQUESTInvalid request body or missing message field
401UNAUTHORIZEDInvalid or expired JWT token
403AGENT_NOT_ENABLEDAgent exists but is not enabled for API access
404AGENT_NOT_FOUNDNo agent found with the given name
429RATE_LIMITEDToo many requests. Retry after the duration in the Retry-After header
500AGENT_EXECUTION_ERRORAgent encountered an internal error during execution