Orchestration Tools
Delegate work to another agent immediately or schedule it to run later.
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 knowledge queries or other application logic.
from connic.tools import trigger_agent, trigger_agent_atfrom connic.tools import trigger_agent, query_knowledge
async def research_and_summarize(topic: str) -> dict:
"""Research a topic and return a summary."""
knowledge = await query_knowledge(
query=f"Information about {topic}",
max_results=5
)
context = "\n".join(
result["content"] for result in knowledge.get("results", [])
)
result = await trigger_agent(
agent_name="researcher",
payload={"topic": topic, "context": context}
)
return {
"topic": topic,
"summary": result["response"],
"sources": len(knowledge.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: gemini/gemini-2.5-pro
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 | string | required | Data sent to the agent as JSON or plain text, for example '{"text": "..."}'. For a tool agent, it becomes the tool's payload argument. |
| 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, 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 runtime recognizes a file dictionary in the payload, the same shape used by 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.
Base64 increases payload size by roughly one third. For files larger than a few megabytes, pass a presigned S3 URL, knowledge entry ID, or database document ID and let the receiving agent fetch the binary.
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: gemini/gemini-2.5-pro
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 | string | required | JSON or plain text sent to the target. For a tool agent it becomes the tool's payload argument. |
| 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
)