Orchestration Tools
Delegate work to another agent immediately or schedule it to run later.
On this page
Using predefined tools in custom tools
Predefined tools can be called by an LLM through its YAML tool list or imported directly into a custom Python tool. Direct imports let you combine orchestration with retrieval queries or other application logic.
from connic.tools import trigger_agent, trigger_agent_atfrom connic.tools import trigger_agent, retrieval_query
async def research_and_summarize(topic: str) -> dict:
"""Research a topic and return a summary."""
retrieval_results = await retrieval_query(
query=f"Information about {topic}",
max_results=5
)
context = "\n".join(
result["content"] for result in retrieval_results.get("results", [])
)
result = await trigger_agent(
agent_name="researcher",
payload={"topic": topic, "context": context}
)
return {
"topic": topic,
"summary": result["response"],
"sources": len(retrieval_results.get("results", []))
}tools:
- orchestration.research_and_summarizetrigger_agent
Orchestrate multiple agents from a single agent
trigger_agent lets one agent call another agent in the same project. Use it to build pipelines, delegate specialized tasks, or coordinate multi-agent workflows.
version: "1.0"
name: orchestrator
model: connic/minimax-m2.7
description: "Coordinates other agents"
system_prompt: |
You orchestrate tasks by delegating to specialized agents.
tools:
- trigger_agentParameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| agent_name | string | required | Name of the agent to trigger |
| payload | dict | list | string | required | Data sent to the agent. For a tool agent, structured JSON text is decoded first, then a dict is passed unchanged as the tool's payload argument; every other value is wrapped as {"input": value}. |
| wait_for_response | bool | true | Wait for completion and include the agent response |
| timeout_seconds | int | 60 | Maximum wait when wait_for_response is true |
Return value
Returns run_id, status (completed, failed, cancelled, awaiting_approval, or timeout),response, and error when applicable. With wait_for_response=False, only run_id is returned.
trigger_agent has no separate files parameter. Instead, the receiving agent accepts a file dictionary in the payload, using the same shape as inbound multipart webhooks and Telegram media.
import base64
from connic.tools import trigger_agent
# Read a file and base64-encode it
with open("/tmp/invoice.pdf", "rb") as f:
encoded = base64.b64encode(f.read()).decode("ascii")
result = await trigger_agent(
agent_name="invoice-extractor",
payload={
"message": "Extract totals from this invoice.",
"files": [
{
"data": encoded,
"mime_type": "application/pdf",
"name": "invoice.pdf",
},
],
"customer_id": "cus_123",
},
wait_for_response=True,
)- Each
filesentry needs base64data,mime_type, andname. - Every non-files key remains at the top level of
context["payload"]. - A
{message, files}payload becomes the plain message for the LLM; richer payloads are serialized as JSON.
trigger_agent_at
Schedule a future agent run with a delay or timestamp
trigger_agent_at schedules another agent for delayed follow-ups, reports, retry-after patterns, and other time-based workflows. The run is created immediately with a scheduled status and dispatched automatically at the scheduled time.
version: "1.0"
name: scheduler
model: connic/minimax-m2.7
description: "Schedules tasks for future execution"
system_prompt: |
You schedule tasks by triggering agents at specific times.
tools:
- trigger_agent_atParameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| agent_name | string | required | Name of the agent to trigger |
| payload | dict | list | string | required | Data sent to the target. For a tool agent, structured JSON text is decoded first, then a dict is passed unchanged as its payload argument; every other value is wrapped as {"input": value}. |
| delay | dict | None | Relative offset using d, h, m, and s. At least one key is required. |
| unix_timestamp | float | None | Absolute Unix timestamp in seconds |
Provide exactly one of delay or unix_timestamp. The maximum scheduling window is 30 days.
Return value
Returns run_id, scheduled_at as an ISO 8601 UTC timestamp, and status: scheduled. It returns immediately without waiting for execution.
Examples
Using a relative delay:
# Schedule a report in 2 hours and 30 minutes
result = await trigger_agent_at(
agent_name="report-generator",
payload={"report_type": "daily"},
delay={"h": 2, "m": 30}
)
# Returns: {"run_id": "...", "scheduled_at": "2026-03-18T16:30:00+00:00", "status": "scheduled"}Using an absolute timestamp:
# Schedule at a specific time (Unix timestamp)
import time
target_time = time.time() + 86400 # 24 hours from now
result = await trigger_agent_at(
agent_name="cleanup-agent",
payload={"scope": "all"},
unix_timestamp=target_time
)