docs-mintlify/recipes/ai/agent-to-agent.mdx
In this recipe, you will learn how to wrap the Cube Chat API as a tool for an external AI agent, enabling agent-to-agent analytics workflows.
When building AI-powered applications, you often have an orchestrating agent (built with frameworks like LangChain, LlamaIndex, or CrewAI) that handles user conversations and coordinates multiple capabilities. One of these capabilities might be answering data questions — revenue trends, customer metrics, pipeline analysis, and so on.
Rather than building a custom data retrieval pipeline, you can give your agent a tool that calls the Cube Chat API. This way, the Cube AI agent handles the hard parts — understanding the data model, writing correct queries, and summarizing results — while your orchestrating agent decides when to ask data questions and how to fold the answers into its broader workflow.
The following diagram shows how the orchestrating agent delegates data questions to the Cube AI agent via the Chat API:
sequenceDiagram
participant User
participant Agent as Your Agent
(LangChain, etc.)
participant Tool as Cube Chat API Tool
participant Cube as Cube AI Agent
User->>Agent: "Prepare a board report
with last quarter financials"
activate Agent
Note over Agent: Decides it needs
financial data
Agent->>Tool: ask_cube("What was total revenue
last quarter, broken down
by product line?")
activate Tool
Tool->>Cube: POST /chat/stream-chat-state
activate Cube
Note over Cube: Queries data model,
runs SQL, summarizes
Cube-->>Tool: Streamed NDJSON response
deactivate Cube
Tool-->>Agent: "Total revenue was $4.2M…"
deactivate Tool
Note over Agent: Incorporates data
into the report
Agent-->>User: Board report with
financial analysis
deactivate Agent
Key benefits of this approach:
chatId, the Cube agent retains
conversational context, so follow-up questions like "now break that
down by region" work automatically.Before you begin, make sure you have:
The core idea is to write a function that sends a question to the Cube Chat API, collects the streamed response, and returns the final answer as a plain string. You then register this function as a tool that your agent can invoke.
Here is a helper that calls the Chat API and extracts the final answer:
import requests
import json
CUBE_CHAT_API_URL = "YOUR_CHAT_API_URL"
CUBE_API_KEY = "YOUR_API_KEY"
def query_cube_agent(question: str, chat_id: str | None = None) -> str:
"""Send a question to the Cube AI agent and return its final answer."""
payload = {
"input": question,
"sessionSettings": {
"externalId": "orchestrating-agent",
},
}
if chat_id:
payload["chatId"] = chat_id
response = requests.post(
CUBE_CHAT_API_URL,
headers={
"Content-Type": "application/json",
"Authorization": f"Api-Key {CUBE_API_KEY}",
},
json=payload,
stream=True,
)
response.raise_for_status()
messages = []
for line in response.iter_lines():
if line:
messages.append(json.loads(line.decode("utf-8")))
# Extract the final answer from the stream
final_messages = [
msg
for msg in messages
if msg.get("role") == "assistant"
and isinstance(msg.get("graphPath"), list)
and len(msg["graphPath"]) > 0
and msg["graphPath"][0] == "final"
and len(msg["graphPath"]) <= 2
]
if final_messages:
return final_messages[-1].get("content", "")
# Fallback: return the last assistant message with content
for msg in reversed(messages):
if msg.get("role") == "assistant" and msg.get("content"):
return msg["content"]
return "No answer received from the Cube agent."
The function filters streamed messages for those where
graphPath[0] === "final" to get the consolidated answer. See the
Chat API reference for details on the response format.
Below is a complete example of a LangChain agent that has access to the
Cube Chat API as a tool. When the agent decides it needs data to answer
a question, it calls the ask_cube tool automatically.
import os
import requests
import json
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
CUBE_CHAT_API_URL = os.environ["CUBE_CHAT_API_URL"]
CUBE_API_KEY = os.environ["CUBE_API_KEY"]
def query_cube_agent(question: str) -> str:
"""Send a question to the Cube AI agent and return its final answer."""
response = requests.post(
CUBE_CHAT_API_URL,
headers={
"Content-Type": "application/json",
"Authorization": f"Api-Key {CUBE_API_KEY}",
},
json={
"input": question,
"sessionSettings": {
"externalId": "orchestrating-agent",
},
},
stream=True,
)
response.raise_for_status()
messages = []
for line in response.iter_lines():
if line:
messages.append(json.loads(line.decode("utf-8")))
final_messages = [
msg
for msg in messages
if msg.get("role") == "assistant"
and isinstance(msg.get("graphPath"), list)
and len(msg["graphPath"]) > 0
and msg["graphPath"][0] == "final"
and len(msg["graphPath"]) <= 2
]
if final_messages:
return final_messages[-1].get("content", "")
for msg in reversed(messages):
if msg.get("role") == "assistant" and msg.get("content"):
return msg["content"]
return "No answer received from the Cube agent."
@tool
def ask_cube(question: str) -> str:
"""Ask a data analytics question. Use this tool whenever you need
business metrics, KPIs, trends, or any data from the company's
databases. Pass a clear, self-contained question."""
return query_cube_agent(question)
llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, [ask_cube])
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
"Prepare a brief executive summary of last quarter's "
"performance. Include revenue, top products, and "
"month-over-month trends."
),
}
]
}
)
print(result["messages"][-1].content)
When you run this, the LangChain agent will:
ask_cube with a focused data question (e.g., "What was total
revenue last quarter?").ask_cube again for additional data points.If your application has per-user data access policies, pass the
current user's identity and attributes through sessionSettings so that
the Cube agent enforces row-level security:
def query_cube_agent_for_user(
question: str,
user_id: str,
user_email: str | None = None,
user_attributes: list[dict] | None = None,
) -> str:
"""Query the Cube agent with user-scoped permissions."""
session_settings = {"externalId": user_id}
if user_email:
session_settings["email"] = user_email
if user_attributes:
session_settings["userAttributes"] = user_attributes
response = requests.post(
CUBE_CHAT_API_URL,
headers={
"Content-Type": "application/json",
"Authorization": f"Api-Key {CUBE_API_KEY}",
},
json={
"input": question,
"sessionSettings": session_settings,
},
stream=True,
)
response.raise_for_status()
# ... same response parsing as above ...
This way, a sales manager asking about revenue will only see data for their territory, while a VP will see the full picture — without any changes to your agent code.
To maintain context across multiple questions in a single workflow, reuse
the chatId returned by the Cube Chat API:
def query_cube_with_followup(questions: list[str]) -> list[str]:
"""Send a sequence of related questions, maintaining conversation context."""
chat_id = None
answers = []
for question in questions:
payload = {
"input": question,
"sessionSettings": {
"externalId": "orchestrating-agent",
},
}
if chat_id:
payload["chatId"] = chat_id
response = requests.post(
CUBE_CHAT_API_URL,
headers={
"Content-Type": "application/json",
"Authorization": f"Api-Key {CUBE_API_KEY}",
},
json=payload,
stream=True,
)
response.raise_for_status()
messages = []
for line in response.iter_lines():
if line:
messages.append(json.loads(line.decode("utf-8")))
# Capture the chatId for follow-up questions
for msg in messages:
if msg.get("id") == "__cutoff__" and msg.get("state", {}).get("chatId"):
chat_id = msg["state"]["chatId"]
final_messages = [
msg
for msg in messages
if msg.get("role") == "assistant"
and isinstance(msg.get("graphPath"), list)
and len(msg["graphPath"]) > 0
and msg["graphPath"][0] == "final"
and len(msg["graphPath"]) <= 2
]
if final_messages:
answers.append(final_messages[-1].get("content", ""))
else:
answers.append("")
return answers
# Example: ask a question and then a follow-up
answers = query_cube_with_followup([
"What was total revenue last quarter?",
"Now break that down by product line.",
])
With this approach, the second question — "Now break that down by product line" — is understood in the context of the first, just like a human conversation.
If your deployment uses multi-agent — with specialized Cube agents like a Sales Assistant and a Marketing Analyst living in the same deployment — your orchestrating agent can choose the right Cube agent for each question and route to it.
The Chat API URL embeds an agentId, so every agent in
your deployment has its own Chat API URL. Copy each one from Admin →
Agents for the corresponding agent. Switching between Cube agents
from your code is then just a matter of POSTing to a different URL —
the request shape is identical.
How your orchestrating agent learns which Cube agents exist is up to you. Two common patterns:
list_cube_agents tool that
returns the available agents at runtime. Useful when agents come and
go, or when you want the orchestrator to pick them up without a code
change.The example below combines both: a list_cube_agents tool for
discovery and a single ask_cube_agent tool that takes an agent name
plus the question and routes the request to the matching Chat API URL.
It builds on the query_cube_agent helper from above, extended to
accept a chat_api_url argument instead of using a single hardcoded
URL.
import json
import os
from langchain_core.tools import tool
CUBE_AGENTS = {
"sales-assistant": {
"url": os.environ["CUBE_SALES_AGENT_URL"],
"description": (
"Sales pipeline, deals, reps, quotas, and revenue by territory."
),
},
"marketing-analyst": {
"url": os.environ["CUBE_MARKETING_AGENT_URL"],
"description": (
"Campaigns, channel attribution, traffic sources, and funnel "
"conversion."
),
},
}
@tool
def list_cube_agents() -> str:
"""List the available Cube agents and the domain each one covers.
Call this first when you need data, then pick the most relevant
agent for the user's question."""
return json.dumps(
{name: agent["description"] for name, agent in CUBE_AGENTS.items()}
)
@tool
def ask_cube_agent(agent_name: str, question: str) -> str:
"""Ask a data analytics question to a specific Cube agent.
Call `list_cube_agents` first to see which agents are available and
which domain each one covers, then pass the chosen `agent_name`
along with a focused, self-contained question."""
agent = CUBE_AGENTS.get(agent_name)
if agent is None:
available = ", ".join(CUBE_AGENTS.keys())
return (
f"Unknown Cube agent '{agent_name}'. "
f"Available agents: {available}."
)
return query_cube_agent(question, chat_api_url=agent["url"])
Alternatively, skip the discovery tool entirely and inline the list of
agents and their domains in your orchestrator's system prompt. The
orchestrator will then pass an agent_name directly to ask_cube_agent
without an extra round trip.
This pattern gives your orchestrating agent a clean routing layer: it picks the right Cube agent for each question, and each Cube agent answers using the rules, certified queries, and accessible views configured for its space. A sales question is routed to the Sales Assistant and answered against the sales views; a marketing question is routed to the Marketing Analyst and answered against the marketing views — without your orchestrator needing to know any of those details.