Chaining in AgenticAPI allows AI agents to orchestrate complex workflows by executing a sequence of action verbs in a single request. This capability, part of the Orchestrate category, enables tasks like fetching data, summarizing it, and notifying a user to be combined seamlessly. The /chain endpoint coordinates these actions, providing traceability, error handling, and support for dynamic workflows. This document explains how chaining works and provides an example implementation.
How Chaining Works #
The /chain endpoint accepts a list of steps, each specifying a verb, path, and parameters. AgenticAPI processes the steps sequentially, returning results for each, including successes and failures. Key features include:
- Traceability: Each step is identified by its verb and parameters, ensuring clear execution tracking.
- Error Handling: Failures are captured per step, allowing partial workflow completion.
- Flexibility: Supports any combination of verbs (e.g.,
FETCH,SUMMARIZE,NOTIFY). - Status Metadata: Indicates step outcomes, guiding agents on next actions.
Chaining is ideal for workflows requiring multiple coordinated actions, such as processing a report or scheduling a meeting.
Example: Chaining FETCH, SUMMARIZE, and NOTIFY #
The following example demonstrates a chain that fetches a document, summarizes it, and notifies a user of the result.
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Dict, Any
app = FastAPI()
class ChainStep(BaseModel):
verb: str
path: str
params: Dict[str, Any]
class ChainRequest(BaseModel):
chain: List[ChainStep]
@app.post("/chain")
async def chain_verbs(chain_request: ChainRequest):
results = []
for step in chain_request.chain:
verb = step.verb.upper()
try:
if verb == "FETCH":
# Simulate fetching a document
result = {"content": "Document text...", "title": "Report"}
elif verb == "SUMMARIZE":
# Simulate summarization
result = {"summary": "Report highlights key trends.", "title": "Report"}
elif verb == "NOTIFY":
# Simulate notification
result = {"status": "Sent", "recipient": step.params["recipient"]}
else:
raise ValueError(f"Unsupported verb: {verb}")
results.append({"step": step.dict(), "result": result})
except Exception as e:
results.append({"step": step.dict(), "error": str(e)})
return {"results": results}
Request:
curl -X POST http://localhost:8000/chain -H "Content-Type: application/json" -d '{
"chain": [
{
"verb": "FETCH",
"path": "/fetch",
"params": {
"document_id": "doc_001",
"output_format": "json"
}
},
{
"verb": "SUMMARIZE",
"path": "/document",
"params": {
"document_id": "doc_001",
"format": "text",
"max_words": 20,
"style": "neutral",
"output_format": "json"
}
},
{
"verb": "NOTIFY",
"path": "/notify",
"params": {
"recipient": "user@example.com",
"message": "Summary completed",
"channel": "email",
"priority": "normal",
"output_format": "json"
}
}
]
}'
Response:
{
"results": [
{
"step": {
"verb": "FETCH",
"path": "/fetch",
"params": {"document_id": "doc_001", "output_format": "json"}
},
"result": {
"content": "Document text...",
"title": "Report"
}
},
{
"step": {
"verb": "SUMMARIZE",
"path": "/document",
"params": {
"document_id": "doc_001",
"format": "text",
"max_words": 20,
"style": "neutral",
"output_format": "json"
}
},
"result": {
"summary": "Report highlights key trends.",
"title": "Report"
}
},
{
"step": {
"verb": "NOTIFY",
"path": "/notify",
"params": {
"recipient": "user@example.com",
"message": "Summary completed",
"channel": "email",
"priority": "normal",
"output_format": "json"
}
},
"result": {
"status": "Sent",
"recipient": "user@example.com"
}
}
]
}
Best Practices #
- Define Clear Steps: Ensure each step specifies a valid verb and parameters.
- Handle Errors: Check for
errorfields in responses to manage failures. - Limit Chain Length: Keep chains concise to maintain performance.
- Use Status Metadata: Leverage step results to guide agent decisions.
Next Steps #
- Learn about Contextual / Dynamic Intent for flexible verbs.
- Explore Responses for output details.
