# Delegate Tasks to Sub-Agents

Delegation gives sub-agents their own context window, tools, and token budget. A sub-agent can be modeled as a separate Inngest function that runs its own [agent loop](/docs-markdown/durable-execution/durable-agents/agent-tool-loops) — the parent either waits for a result or fires and forgets.

## Define a sub-agent function

A sub-agent is a regular Inngest function. It receives a task, runs an agent loop, and returns a result:

```typescript {{ title: "TypeScript" }}
import { inngest } from "./client";
import { createAgentLoop } from "./agent-loop";

export const subAgent = inngest.createFunction(
  { id: "sub-agent", triggers: [{ event: "agent/sub-agent.spawn" }] },
  async ({ event, step, logger }) => {
    const { task, sessionId } = event.data;

    const systemPrompt = `You are a focused sub-agent. Complete the following task and return a clear, concise result.\n\nTask: ${task}`;

    const result = await runAgentLoop({
      step,
      systemPrompt,
      sessionId,
      tools: SUB_AGENT_TOOLS, // No delegation tools — see "Prevent recursion"
      maxIterations: 20,
    });

    return {
      response: result.response,
      iterations: result.iterations,
    };
  }
);
```

```python {{ title: "Python" }}
import typing

import inngest

from .agent_loop import run_agent_loop
from .client import inngest_client
from .tool_sets import SUB_AGENT_TOOLS

@inngest_client.create_function(
    fn_id="sub-agent",
    trigger=inngest.TriggerEvent(event="agent/sub-agent.spawn"),
)
async def sub_agent(ctx: inngest.Context) -> dict[str, typing.Any]:
    task = ctx.event.data["task"]
    session_id = str(ctx.event.data["sessionId"])

    system_prompt = (
        "You are a focused sub-agent. Complete the following task and "
        f"return a clear, concise result.\n\nTask: {task}"
    )

    result = await run_agent_loop(
        ctx,
        system_prompt=system_prompt,
        session_id=session_id,
        # No delegation tools — see "Prevent recursion"
        tools=SUB_AGENT_TOOLS,
        max_iterations=20,
    )

    return {
        "response": result["response"],
        "iterations": result["iterations"],
    }
```

```go {{ title: "Go" }}
import (
	"context"
	"fmt"

	"github.com/inngest/inngestgo"
)

type SubAgentSpawn struct {
	Task      string `json:"task"`
	SessionID string `json:"sessionId"`
}

func SubAgent(client inngestgo.Client) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{ID: "sub-agent"},
		inngestgo.EventTrigger("agent/sub-agent.spawn", nil),
		func(ctx context.Context, input inngestgo.Input[SubAgentSpawn]) (any, error) {
			systemPrompt := fmt.Sprintf("You are a focused sub-agent. Complete the following task and return a clear, concise result.\n\nTask: %s", input.Event.Data.Task)

			result, err := runAgentLoop(ctx, AgentLoopOpts{
				SystemPrompt:  systemPrompt,
				SessionID:     input.Event.Data.SessionID,
				Tools:         SUB_AGENT_TOOLS, // No delegation tools — see "Prevent recursion"
				MaxIterations: 20,
			})
			if err != nil {
				return nil, err
			}

			return AgentResult{
				Response:   result.Response,
				Iterations: result.Iterations,
			}, nil
		},
	)
}
```

The sub-agent uses a **restricted tool set** without delegation tools to prevent infinite recursion ([details below](#prevent-recursion)).

## Define the delegation tool

Give the parent agent's LLM a tool that makes delegation a natural choice:

```typescript {{ title: "TypeScript" }}
const delegateTaskTool = {
  type: "function",
  function: {
    name: "delegate_task",
    description:
      "Delegate a task to a sub-agent that will work on it independently and return a result. " +
      "Use this for tasks that require deep research, many tool calls, or focused work " +
      "that would clutter the current conversation.",
    parameters: {
      type: "object",
      properties: {
        task: {
          type: "string",
          description:
            "A clear, self-contained description of the task. Include all necessary context — " +
            "the sub-agent does not have access to this conversation's history.",
        },
      },
      required: ["task"],
    },
  },
};
```

```python {{ title: "Python" }}
delegate_task_tool = {
    "type": "function",
    "function": {
        "name": "delegate_task",
        "description": (
            "Delegate a task to a sub-agent that will work on it "
            "independently and return a result. Use this for tasks that "
            "require deep research, many tool calls, or focused work that "
            "would clutter the current conversation."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "task": {
                    "type": "string",
                    "description": (
                        "A clear, self-contained description of the task. "
                        "Include all necessary context — the sub-agent "
                        "does not have access to this conversation's "
                        "history."
                    ),
                },
            },
            "required": ["task"],
        },
    },
}
```

```go {{ title: "Go" }}
import "github.com/sashabaranov/go-openai"

var delegateTaskTool = openai.Tool{
	Type: openai.ToolTypeFunction,
	Function: &openai.FunctionDefinition{
		Name: "delegate_task",
		Description: "Delegate a task to a sub-agent that will work on it independently and return a result. " +
			"Use this for tasks that require deep research, many tool calls, or focused work " +
			"that would clutter the current conversation.",
		Parameters: map[string]any{
			"type": "object",
			"properties": map[string]any{
				"task": map[string]any{
					"type": "string",
					"description": "A clear, self-contained description of the task. Include all necessary context — " +
						"the sub-agent does not have access to this conversation's history.",
				},
			},
			"required": []string{"task"},
		},
	},
}
```

## Delegate synchronously with `step.invoke()`

To delegate and wait for a result, use [`step.invoke()`](/docs-markdown/durable-execution/primitives/step-invoke) — the parent blocks until the sub-agent returns:

```typescript {{ title: "TypeScript" }}
import { subAgent } from "./sub-agent";

export const parentAgent = inngest.createFunction(
  { id: "parent-agent", triggers: [{ event: "agent/task.received" }] },
  async ({ event, step }) => {
    let messages = [{ role: "system", content: SYSTEM_PROMPT }];
    let done = false;
    let i = 0;

    while (!done && i < 30) {
      const response = await step.run(`think`, async () => {
        return await callLLM(messages, TOOLS);
      });

      for (const toolCall of response.toolCalls) {
        let toolResult: string;

        if (toolCall.name === "delegate_task") {
          // Synchronous delegation — parent waits for the result
          const subResult = await step.invoke(`sub-agent`, {
            function: subAgent,
            data: {
              task: toolCall.arguments.task,
              sessionId: `sub-${event.data.sessionId}-${Date.now()}`,
            },
          });

          toolResult = subResult?.response ?? "(no response from sub-agent)";
        } else {
          toolResult = await step.run(`tool-${toolCall.name}`, async () => {
            return await executeTool(toolCall.name, toolCall.arguments);
          });
        }

        messages.push(
          { role: "assistant", content: null, tool_calls: [toolCall] },
          { role: "tool", tool_call_id: toolCall.id, content: toolResult }
        );
      }

      if (response.toolCalls.length === 0) {
        done = true;
      }
      i++;
    }

    return { response: messages[messages.length - 1].content };
  }
);
```

```python {{ title: "Python" }}
import time
import typing

import inngest

from .client import inngest_client
from .sub_agent import sub_agent

@inngest_client.create_function(
    fn_id="parent-agent",
    trigger=inngest.TriggerEvent(event="agent/task.received"),
)
async def parent_agent(ctx: inngest.Context) -> dict[str, typing.Any]:
    messages: list[dict[str, typing.Any]] = [
        {"role": "system", "content": SYSTEM_PROMPT}
    ]
    done = False
    i = 0

    while not done and i < 30:
        response = await ctx.step.run("think", call_llm, messages, TOOLS)

        for tool_call in response["tool_calls"]:
            if tool_call["name"] == "delegate_task":
                # Synchronous delegation — parent waits for the result
                session_id = ctx.event.data["sessionId"]
                sub_result = await ctx.step.invoke(
                    "sub-agent",
                    function=sub_agent,
                    data={
                        "task": tool_call["arguments"]["task"],
                        "sessionId": f"sub-{session_id}-{int(time.time() * 1000)}",
                    },
                )

                tool_result = str(
                    (sub_result or {}).get("response")
                    or "(no response from sub-agent)"
                )
            else:
                tool_result = await ctx.step.run(
                    f"tool-{tool_call['name']}",
                    execute_tool,
                    tool_call["name"],
                    tool_call["arguments"],
                )

            messages.extend(
                [
                    {
                        "role": "assistant",
                        "content": None,
                        "tool_calls": [tool_call],
                    },
                    {
                        "role": "tool",
                        "tool_call_id": tool_call["id"],
                        "content": tool_result,
                    },
                ]
            )

        if not response["tool_calls"]:
            done = True
        i += 1

    return {"response": messages[-1]["content"]}
```

```go {{ title: "Go" }}
import (
	"context"
	"fmt"
	"time"

	"github.com/inngest/inngestgo"
	"github.com/inngest/inngestgo/step"
)

type TaskReceived struct {
	SessionID string `json:"sessionId"`
}

func ParentAgent(client inngestgo.Client, subAgent inngestgo.ServableFunction) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{ID: "parent-agent"},
		inngestgo.EventTrigger("agent/task.received", nil),
		func(ctx context.Context, input inngestgo.Input[TaskReceived]) (any, error) {
			messages := []Message{{Role: "system", Content: SYSTEM_PROMPT}}
			done := false
			i := 0

			for !done && i < 30 {
				response, err := step.Run(ctx, "think", func(ctx context.Context) (LLMResponse, error) {
					return callLLM(ctx, messages, TOOLS)
				})
				if err != nil {
					return nil, err
				}

				for _, toolCall := range response.ToolCalls {
					var toolResult string

					if toolCall.Name == "delegate_task" {
						// Synchronous delegation — parent waits for the result
						subResult, err := step.Invoke[AgentResult](ctx, "sub-agent", step.InvokeOpts{
							FunctionId: subAgent.FullyQualifiedID(),
							Data: map[string]any{
								"task":      toolCall.Arguments["task"],
								"sessionId": fmt.Sprintf("sub-%s-%d", input.Event.Data.SessionID, time.Now().UnixMilli()),
							},
						})
						if err != nil {
							return nil, err
						}

						toolResult = subResult.Response
						if toolResult == "" {
							toolResult = "(no response from sub-agent)"
						}
					} else {
						toolResult, err = step.Run(ctx, "tool-"+toolCall.Name, func(ctx context.Context) (string, error) {
							return executeTool(ctx, toolCall.Name, toolCall.Arguments)
						})
						if err != nil {
							return nil, err
						}
					}

					messages = append(messages,
						Message{Role: "assistant", ToolCalls: []ToolCall{toolCall}},
						Message{Role: "tool", ToolCallID: toolCall.ID, Content: toolResult},
					)
				}

				if len(response.ToolCalls) == 0 {
					done = true
				}
				i++
			}

			return map[string]any{"response": messages[len(messages)-1].Content}, nil
		},
	)
}
```

Because `step.invoke()` is a durable step, the parent pauses execution and resumes exactly where it left off when the sub-agent completes. The parent's LLM sees only the sub-agent's summary, not its full internal conversation.

## Delegate asynchronously with `step.sendEvent()`

To delegate without waiting, use [`step.sendEvent()`](/docs-markdown/durable-execution/primitives/step-sendevent) — the parent fires an event and continues:

```typescript {{ title: "TypeScript" }}
if (toolCall.name === "delegate_background_task") {
  await step.sendEvent("spawn-background-task", {
    name: "agent/sub-agent.spawn",
    data: {
      task: toolCall.arguments.task,
      sessionId: `sub-${event.data.sessionId}-${Date.now()}`,
      isAsync: true,
      replyTo: {
        type: "webhook",
        url: event.data.callbackUrl,
      },
    },
  });

  toolResult = "Task delegated. The sub-agent is working on it in the background.";
}
```

```python {{ title: "Python" }}
if tool_call["name"] == "delegate_background_task":
    session_id = ctx.event.data["sessionId"]
    await ctx.step.send_event(
        "spawn-background-task",
        inngest.Event(
            name="agent/sub-agent.spawn",
            data={
                "task": tool_call["arguments"]["task"],
                "sessionId": f"sub-{session_id}-{int(time.time() * 1000)}",
                "isAsync": True,
                "replyTo": {
                    "type": "webhook",
                    "url": ctx.event.data["callbackUrl"],
                },
            },
        ),
    )

    tool_result = (
        "Task delegated. The sub-agent is working on it in the background."
    )
```

```go {{ title: "Go" }}
if toolCall.Name == "delegate_background_task" {
	_, err := step.Send(ctx, "spawn-background-task", inngestgo.Event{
		Name: "agent/sub-agent.spawn",
		Data: map[string]any{
			"task":      toolCall.Arguments["task"],
			"sessionId": fmt.Sprintf("sub-%s-%d", input.Event.Data.SessionID, time.Now().UnixMilli()),
			"isAsync":   true,
			"replyTo": map[string]any{
				"type": "webhook",
				"url":  input.Event.Data.CallbackURL,
			},
		},
	})
	if err != nil {
		return "", err
	}

	toolResult = "Task delegated. The sub-agent is working on it in the background."
}
```

To deliver results when the sub-agent finishes, handle async mode in the sub-agent:

```typescript {{ title: "TypeScript" }}
export const subAgent = inngest.createFunction(
  { id: "sub-agent", triggers: [{ event: "agent/sub-agent.spawn" }] },
  async ({ event, step, logger }) => {
    const { task, sessionId, isAsync, replyTo } = event.data;

    const result = await runAgentLoop({
      step,
      systemPrompt: `Complete this task:\n\n${task}`,
      sessionId,
      tools: SUB_AGENT_TOOLS,
      maxIterations: 30,
    });

    if (isAsync && replyTo) {
      await step.run("deliver-result", async () => {
        if (replyTo.type === "webhook") {
          await fetch(replyTo.url, {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ response: result.response }),
          });
        }
      });
    }

    return result;
  }
);
```

```python {{ title: "Python" }}
import typing

import httpx
import inngest

from .agent_loop import run_agent_loop
from .client import inngest_client
from .tool_sets import SUB_AGENT_TOOLS

@inngest_client.create_function(
    fn_id="sub-agent",
    trigger=inngest.TriggerEvent(event="agent/sub-agent.spawn"),
)
async def sub_agent(ctx: inngest.Context) -> dict[str, typing.Any]:
    task = ctx.event.data["task"]
    is_async = ctx.event.data.get("isAsync")
    reply_to = ctx.event.data.get("replyTo")

    result = await run_agent_loop(
        ctx,
        system_prompt=f"Complete this task:\n\n{task}",
        session_id=str(ctx.event.data["sessionId"]),
        tools=SUB_AGENT_TOOLS,
        max_iterations=30,
    )

    if is_async and isinstance(reply_to, dict):

        async def deliver_result() -> None:
            if reply_to["type"] == "webhook":
                async with httpx.AsyncClient() as http:
                    res = await http.post(
                        str(reply_to["url"]),
                        json={"response": result["response"]},
                    )
                    res.raise_for_status()

        await ctx.step.run("deliver-result", deliver_result)

    return result
```

```go {{ title: "Go" }}
import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"

	"github.com/inngest/inngestgo"
	"github.com/inngest/inngestgo/step"
)

type AsyncSubAgentSpawn struct {
	Task      string `json:"task"`
	SessionID string `json:"sessionId"`
	IsAsync   bool   `json:"isAsync"`
	ReplyTo   *struct {
		Type string `json:"type"`
		URL  string `json:"url"`
	} `json:"replyTo"`
}

func SubAgentAsync(client inngestgo.Client) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{ID: "sub-agent"},
		inngestgo.EventTrigger("agent/sub-agent.spawn", nil),
		func(ctx context.Context, input inngestgo.Input[AsyncSubAgentSpawn]) (any, error) {
			data := input.Event.Data

			result, err := runAgentLoop(ctx, AgentLoopOpts{
				SystemPrompt:  fmt.Sprintf("Complete this task:\n\n%s", data.Task),
				SessionID:     data.SessionID,
				Tools:         SUB_AGENT_TOOLS,
				MaxIterations: 30,
			})
			if err != nil {
				return nil, err
			}

			if data.IsAsync && data.ReplyTo != nil {
				_, err := step.Run(ctx, "deliver-result", func(ctx context.Context) (any, error) {
					if data.ReplyTo.Type != "webhook" {
						return nil, nil
					}
					body, _ := json.Marshal(map[string]any{"response": result.Response})
					req, err := http.NewRequestWithContext(ctx, http.MethodPost, data.ReplyTo.URL, bytes.NewReader(body))
					if err != nil {
						return nil, err
					}
					req.Header.Set("Content-Type", "application/json")
					resp, err := http.DefaultClient.Do(req)
					if err != nil {
						return nil, err
					}
					return nil, resp.Body.Close()
				})
				if err != nil {
					return nil, err
				}
			}

			return result, nil
		},
	)
}
```

Alternatively, the sub-agent can emit a result event and a separate function handles delivery — this decouples the sub-agent from routing logic:

```typescript {{ title: "TypeScript" }}
// Sub-agent emits result as an event
if (isAsync) {
  await step.sendEvent("result-ready", {
    name: "agent/sub-agent.completed",
    data: {
      sessionId,
      parentSessionId: event.data.parentSessionId,
      response: result.response,
    },
  });
}
```

```python {{ title: "Python" }}
# Sub-agent emits result as an event
if is_async:
    await ctx.step.send_event(
        "result-ready",
        inngest.Event(
            name="agent/sub-agent.completed",
            data={
                "sessionId": session_id,
                "parentSessionId": ctx.event.data["parentSessionId"],
                "response": result["response"],
            },
        ),
    )
```

```go {{ title: "Go" }}
// Sub-agent emits result as an event
if isAsync {
	_, err := step.Send(ctx, "result-ready", inngestgo.Event{
		Name: "agent/sub-agent.completed",
		Data: map[string]any{
			"sessionId":       sessionID,
			"parentSessionId": input.Event.Data.ParentSessionID,
			"response":        result.Response,
		},
	})
	if err != nil {
		return err
	}
}
```

```typescript {{ title: "TypeScript" }}
// Separate function handles result delivery
export const deliverSubAgentResult = inngest.createFunction(
  { id: "deliver-sub-agent-result", triggers: { event: "agent/sub-agent.completed" } },
  async ({ event, step }) => {
    const { response, parentSessionId } = event.data;

    await step.run("deliver", async () => {
      await notifyUser(parentSessionId, response);
    });
  }
);
```

```python {{ title: "Python" }}
import inngest

from .client import inngest_client

# Separate function handles result delivery
@inngest_client.create_function(
    fn_id="deliver-sub-agent-result",
    trigger=inngest.TriggerEvent(event="agent/sub-agent.completed"),
)
async def deliver_sub_agent_result(ctx: inngest.Context) -> None:
    response = str(ctx.event.data["response"])
    parent_session_id = str(ctx.event.data["parentSessionId"])

    await ctx.step.run("deliver", notify_user, parent_session_id, response)
```

```go {{ title: "Go" }}
import (
	"context"

	"github.com/inngest/inngestgo"
	"github.com/inngest/inngestgo/step"
)

type SubAgentCompleted struct {
	Response        string `json:"response"`
	ParentSessionID string `json:"parentSessionId"`
}

// Separate function handles result delivery
func DeliverSubAgentResult(client inngestgo.Client) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{ID: "deliver-sub-agent-result"},
		inngestgo.EventTrigger("agent/sub-agent.completed", nil),
		func(ctx context.Context, input inngestgo.Input[SubAgentCompleted]) (any, error) {
			_, err := step.Run(ctx, "deliver", func(ctx context.Context) (any, error) {
				return nil, notifyUser(ctx, input.Event.Data.ParentSessionID, input.Event.Data.Response)
			})
			return nil, err
		},
	)
}
```

### Choose sync vs. async

|                     | Sync (`step.invoke()`)                                                            | Async (`step.sendEvent()`)                                                          |
| ------------------- | --------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------- |
| **Parent blocks?**  | Yes — waits for result                                                            | No — continues immediately                                                          |
| **Result flows to** | Parent agent's tool output                                                        | Separate delivery (webhook, event, notification)                                    |
| **Best for**        | Tasks where the parent needs the answer to continue (research, lookups, analysis) | Long-running tasks where the user can be notified later (reports, batch processing) |
| **Timeout**         | Subject to function execution time limits                                         | Sub-agent runs on its own timeline                                                  |
| **Retry behavior**  | If sub-agent fails, parent's step retries                                         | Sub-agent retries independently                                                     |

## Prevent recursion

If a sub-agent has delegation tools, it could spawn sub-agents indefinitely. Prevent this by restricting the tool set:

```typescript {{ title: "TypeScript" }}
// Tools available to the parent agent
const PARENT_TOOLS = [
  searchTool,
  readFileTool,
  writeFileTool,
  delegateTaskTool,        // Can delegate
  delegateBackgroundTool,  // Can delegate async
];

// Tools available to sub-agents — no delegation
const SUB_AGENT_TOOLS = [
  searchTool,
  readFileTool,
  writeFileTool,
  // No delegate tools — sub-agents cannot spawn further sub-agents
];
```

```python {{ title: "Python" }}
# Tools available to the parent agent
PARENT_TOOLS = [
    search_tool,
    read_file_tool,
    write_file_tool,
    delegate_task_tool,  # Can delegate
    delegate_background_tool,  # Can delegate async
]

# Tools available to sub-agents — no delegation
SUB_AGENT_TOOLS = [
    search_tool,
    read_file_tool,
    write_file_tool,
    # No delegate tools — sub-agents cannot spawn further sub-agents
]
```

```go {{ title: "Go" }}
// Tools available to the parent agent
var PARENT_TOOLS = []openai.Tool{
	searchTool,
	readFileTool,
	writeFileTool,
	delegateTaskTool,       // Can delegate
	delegateBackgroundTool, // Can delegate async
}

// Tools available to sub-agents — no delegation
var SUB_AGENT_TOOLS = []openai.Tool{
	searchTool,
	readFileTool,
	writeFileTool,
	// No delegate tools — sub-agents cannot spawn further sub-agents
}
```

Combine tool restriction with a hard iteration cap for two layers of protection:

```typescript {{ title: "TypeScript" }}
export const subAgent = inngest.createFunction(
  { id: "sub-agent", retries: 1, triggers: { event: "agent/sub-agent.spawn" } },
  async ({ event, step }) => {
    const result = await runAgentLoop({
      step,
      systemPrompt: `Complete this task:\n\n${event.data.task}`,
      sessionId: event.data.sessionId,
      tools: SUB_AGENT_TOOLS, // Restricted — always
      maxIterations: 20,      // Hard cap on iterations
    });

    return result;
  }
);
```

```python {{ title: "Python" }}
import typing

import inngest

from .agent_loop import run_agent_loop
from .client import inngest_client
from .tool_sets import SUB_AGENT_TOOLS

@inngest_client.create_function(
    fn_id="sub-agent",
    retries=1,
    trigger=inngest.TriggerEvent(event="agent/sub-agent.spawn"),
)
async def sub_agent(ctx: inngest.Context) -> dict[str, typing.Any]:
    result = await run_agent_loop(
        ctx,
        system_prompt=f"Complete this task:\n\n{ctx.event.data['task']}",
        session_id=str(ctx.event.data["sessionId"]),
        tools=SUB_AGENT_TOOLS,  # Restricted — always
        max_iterations=20,  # Hard cap on iterations
    )

    return result
```

```go {{ title: "Go" }}
import (
	"context"
	"fmt"

	"github.com/inngest/inngestgo"
)

func SubAgentWithLimits(client inngestgo.Client) (inngestgo.ServableFunction, error) {
	return inngestgo.CreateFunction(
		client,
		inngestgo.FunctionOpts{ID: "sub-agent", Retries: inngestgo.IntPtr(1)},
		inngestgo.EventTrigger("agent/sub-agent.spawn", nil),
		func(ctx context.Context, input inngestgo.Input[SubAgentSpawn]) (any, error) {
			return runAgentLoop(ctx, AgentLoopOpts{
				SystemPrompt:  fmt.Sprintf("Complete this task:\n\n%s", input.Event.Data.Task),
				SessionID:     input.Event.Data.SessionID,
				Tools:         SUB_AGENT_TOOLS, // Restricted — always
				MaxIterations: 20,              // Hard cap on iterations
			})
		},
	)
}
```

## Handle sub-agent failures

`step.invoke()` retries the sub-agent based on its `retries` config. If all retries are exhausted, the error propagates to the parent. To let the parent LLM adapt, catch the error:

```typescript {{ title: "TypeScript" }}
let toolResult: string;
try {
  const subResult = await step.invoke(`sub-agent`, {
    function: subAgent,
    data: { task: toolCall.arguments.task, sessionId: subSessionId },
  });
  toolResult = subResult?.response ?? "(no response)";
} catch (error) {
  toolResult = `Sub-agent failed: ${error.message}. You may need to handle this task directly.`;
}
```

```python {{ title: "Python" }}
try:
    sub_result = await ctx.step.invoke(
        "sub-agent",
        function=sub_agent,
        data={
            "task": tool_call["arguments"]["task"],
            "sessionId": sub_session_id,
        },
    )
    tool_result = str(
        (sub_result or {}).get("response") or "(no response)"
    )
except inngest.StepError as err:
    tool_result = (
        f"Sub-agent failed: {err.message}. "
        "You may need to handle this task directly."
    )
```

```go {{ title: "Go" }}
var toolResult string
subResult, err := step.Invoke[AgentResult](ctx, "sub-agent", step.InvokeOpts{
	FunctionId: subAgent.FullyQualifiedID(),
	Data:       map[string]any{"task": toolCall.Arguments["task"], "sessionId": subSessionID},
})
if err != nil {
	toolResult = fmt.Sprintf("Sub-agent failed: %s. You may need to handle this task directly.", err)
} else if subResult.Response != "" {
	toolResult = subResult.Response
} else {
	toolResult = "(no response)"
}
```

## Schedule a sub-agent

To run a sub-agent at a future time, include a [timestamp](/docs-markdown/durable-execution/guides-and-advanced/events-and-triggers/schedules-and-delayed-starts#start-once-in-the-future) (`ts`; `Timestamp` in Go) when sending the event:

```typescript {{ title: "TypeScript" }}
await step.sendEvent("schedule-daily-report", {
  name: "agent/sub-agent.spawn",
  data: {
    task: "Generate the daily analytics summary report.",
    sessionId: `scheduled-${Date.now()}`,
    isAsync: true,
    replyTo: { type: "webhook", url: REPORT_WEBHOOK_URL },
  },
  ts: tomorrow9am.getTime(),
});
```

```python {{ title: "Python" }}
await ctx.step.send_event(
    "schedule-daily-report",
    inngest.Event(
        name="agent/sub-agent.spawn",
        data={
            "task": "Generate the daily analytics summary report.",
            "sessionId": f"scheduled-{int(time.time() * 1000)}",
            "isAsync": True,
            "replyTo": {"type": "webhook", "url": REPORT_WEBHOOK_URL},
        },
        ts=int(tomorrow_9am.timestamp() * 1000),
    ),
)
```

```go {{ title: "Go" }}
_, err := step.Send(ctx, "schedule-daily-report", inngestgo.Event{
	Name: "agent/sub-agent.spawn",
	Data: map[string]any{
		"task":      "Generate the daily analytics summary report.",
		"sessionId": fmt.Sprintf("scheduled-%d", time.Now().UnixMilli()),
		"isAsync":   true,
		"replyTo":   map[string]any{"type": "webhook", "url": REPORT_WEBHOOK_URL},
	},
	Timestamp: tomorrow9am.UnixMilli(),
})
```

For recurring work, use a [cron-triggered function](/docs-markdown/durable-execution/guides-and-advanced/events-and-triggers/schedules-and-delayed-starts#run-on-a-recurring-schedule) instead.

## Write self-contained task descriptions

In this set up, you can choose how context is shared between parent and sub-agents. With the above approach, the sub-agent gets it's main context from the "task" given to the sub-agent. In this approach, you'll want to include everything it needs in the task itself:

```typescript {{ title: "TypeScript" }}
// ❌ Bad — relies on context the sub-agent doesn't have
{ task: "Summarize what we discussed above" }

// ✅ Good — self-contained with all necessary context
{ task: "Summarize the key findings from the Q4 2025 revenue report. Focus on: 1) YoY growth rate, 2) top performing segments, 3) areas of concern." }
```

```python {{ title: "Python" }}
# ❌ Bad — relies on context the sub-agent doesn't have
bad = {"task": "Summarize what we discussed above"}

# ✅ Good — self-contained with all necessary context
good = {
    "task": (
        "Summarize the key findings from the Q4 2025 revenue report. "
        "Focus on: 1) YoY growth rate, 2) top performing segments, "
        "3) areas of concern."
    )
}
```

```go {{ title: "Go" }}
// ❌ Bad — relies on context the sub-agent doesn't have
bad := map[string]any{"task": "Summarize what we discussed above"}

// ✅ Good — self-contained with all necessary context
good := map[string]any{"task": "Summarize the key findings from the Q4 2025 revenue report. Focus on: 1) YoY growth rate, 2) top performing segments, 3) areas of concern."}
```

## Generic vs. specialized sub-agents

Generic sub-agents are a great way to get started and work for many use cases. If your system requires more specialized sets of tools or different models for sub-agents, you might consider creating specialized sub-agents.

To create a system with specialized sub-agents, follow the patterns above, but create multiple tools, with each that invoke their own agent or a "loader" sub-agent that can handle multiple agent types conditionally. As a recommendation, LLMs often do better with separate tools for separate sub-agents rather than a single tool with different parameters for selection.

The task description and available tools are enough to specialize behavior.

## Next steps

- [Build an Agent Tool Loop](/docs-markdown/durable-execution/durable-agents/agent-tool-loops) — Build the agent loop that powers both parent and sub-agents.
- [Pause for Human Approval](/docs-markdown/durable-execution/durable-agents/human-in-the-loop) — Add approval gates before or after delegation with `step.waitForEvent()`.
- [Three sub-agent patterns you need for your agentic system](/blog/three-patterns-you-need-for-agentic-systems) — sync, async, and scheduled delegation patterns in depth