Orchestration Tools
Agenten können Arbeit sofort an andere Agenten delegieren oder die Ausführung für einen späteren Zeitpunkt planen.
Auf dieser Seite
Vordefinierte Tools in Custom Tools verwenden
Vordefinierte Tools können von einem LLM über seine YAML-Tool-Liste aufgerufen oder direkt in ein Custom Python Tool importiert werden. Direkte Imports verbinden Orchestration mit Retrieval Queries oder anderer Anwendungslogik.
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
Mehrere Agenten von einem einzelnen Agenten aus orchestrieren
trigger_agent lässt einen Agenten einen anderen Agenten im selben Projekt aufrufen. Verwende es für Pipelines, spezialisierte delegierte Aufgaben oder 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_agentParameter
| Parameter | Typ | Standard | Beschreibung |
|---|---|---|---|
| agent_name | string | erforderlich | Name des zu startenden Agenten |
| payload | dict | list | string | erforderlich | Daten für den Agenten. Bei einem Tool-Agenten wird strukturierter JSON-Text zuerst dekodiert. Ein Dictionary wird dann unverändert als Argument payload an das Tool übergeben; jeder andere Wert wird als {"input": value} verpackt. |
| wait_for_response | bool | true | Auf den Abschluss warten und die Antwort des Agenten zurückgeben |
| timeout_seconds | int | 60 | Maximale Wartezeit, wenn wait_for_response true ist |
Rückgabewert
Gibt run_id, status (completed, failed, cancelled, awaiting_approval oder timeout), response und gegebenenfalls error zurück. Mit wait_for_response=False wird nur run_id zurückgegeben.
trigger_agent hat keinen separaten Parameter files. Stattdessen akzeptiert der empfangende Agent ein Datei-Dictionary in der Payload, in derselben Struktur wie Inbound-Multipart-Webhooks und Telegram-Medien.
import base64
from connic.tools import trigger_agent
# Datei lesen und Base64-codieren
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,
)- Jeder
files-Eintrag benötigt Base64-data,mime_typeundname. - Jeder Schlüssel außer files bleibt auf der obersten Ebene von
context["payload"]. - Eine
{message, files}-Payload wird zur einfachen Nachricht für das LLM; umfangreichere Payloads werden als JSON serialisiert.
trigger_agent_at
Einen künftigen Agenten-Run mit Verzögerung oder Zeitstempel planen
trigger_agent_at plant einen anderen Agenten für verzögerte Follow-ups, Berichte, Retry-After-Muster und andere zeitbasierte Workflows. Der Run wird sofort mit dem Status scheduled erstellt und zum geplanten Zeitpunkt automatisch gestartet.
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_atParameter
| Parameter | Typ | Standard | Beschreibung |
|---|---|---|---|
| agent_name | string | erforderlich | Name des zu startenden Agenten |
| payload | dict | list | string | erforderlich | Daten für das Ziel. Bei einem Tool-Agenten wird strukturierter JSON-Text zuerst dekodiert. Ein Dictionary wird dann unverändert als Payload-Argument übergeben; jeder andere Wert wird als {"input": value} verpackt. |
| delay | dict | None | Wartezeit bis zur Ausführung, angegeben mit d, h, m und s. Mindestens ein Schlüssel ist erforderlich. |
| unix_timestamp | float | None | Absoluter Unix-Zeitstempel in Sekunden |
Gib genau einen der Werte delay oder unix_timestamp an. Der maximale Planungszeitraum beträgt 30 Tage.
Rückgabewert
Gibt run_id, scheduled_at als ISO-8601-UTC-Zeitstempel und status: scheduled zurück. Die Funktion kehrt sofort zurück, ohne auf die Ausführung zu warten.
Beispiele
Mit einer relativen Verzögerung:
# Report für 2 Stunden und 30 Minuten später planen
result = await trigger_agent_at(
agent_name="report-generator",
payload={"report_type": "daily"},
delay={"h": 2, "m": 30}
)
# Rückgabe: {"run_id": "...", "scheduled_at": "2026-03-18T16:30:00+00:00", "status": "scheduled"}Mit einem absoluten Zeitstempel:
# Für einen bestimmten Zeitpunkt planen (Unix-Zeitstempel)
import time
target_time = time.time() + 86400 # in 24 Stunden
result = await trigger_agent_at(
agent_name="cleanup-agent",
payload={"scope": "all"},
unix_timestamp=target_time
)