Run work in the background

Some operations should not block the agent’s response. Sending notifications, writing to a slow store, triggering webhooks, or updating a database are good candidates to run in the background. BackgroundTaskManager launches these tasks asynchronously from inside any node function, returning control to the caller immediately.

When to use background tasks

Background tasks are ideal for fire-and-forget operations that do not affect the current response:

• Sending notifications (email, SMS, push notifications) after a successful action
• Logging to external systems or audit trails
• Triggering webhooks for downstream subscribers
• Uploading large files to storage (S3, GCS, etc.)
• Writing to a slow or non-critical database
• Cleanup or maintenance tasks (clearing old files, archiving data)
• Metrics and analytics reporting

The key characteristic: the user or the next graph step does not need to wait for the task to finish.

When NOT to use background tasks

Do not use background tasks when:

• The result is needed by the current response or later steps in the graph
• The operation affects data the agent depends on (e.g., updating a vector store the agent queries)
• You need to know if the task succeeded before proceeding
• The operation is critical and failure should block the entire flow

In these cases, process the work synchronously in your node before returning state.

Prerequisites

You have a working graph. The tenxgraph package is installed. BackgroundTaskManager is automatically available in every node via dependency injection; no extra configuration is needed.

Quick start

Declare task_manager: BackgroundTaskManager = Inject[BackgroundTaskManager] as a parameter in your node function. The framework injects it automatically at runtime.

import asyncio
from tenxgraph.core import StateGraph
from tenxgraph.core.state import AgentState, Message
from tenxgraph.utils import END
from tenxgraph.utils.background_task_manager import BackgroundTaskManager
from injectq import Inject

async def send_notification(user_id: str, text: str) -> None:
    """Simulate sending a push notification (slow I/O)."""
    await asyncio.sleep(0.5)
    print(f"Notification sent to {user_id}: {text}")

async def my_node(
    state: AgentState,
    config: dict,
    task_manager: BackgroundTaskManager = Inject[BackgroundTaskManager],
) -> Message:
    # Do main work and return immediately
    reply = Message.text_message(
        "Your report is being processed in the background.", role="assistant"
    )

    # Fire-and-forget: doesn't block the response
    task_manager.create_task(
        send_notification(config.get("user_id", "anon"), "Report ready soon"),
        name="send_notification",
        timeout=10.0,
    )

    return reply

graph = StateGraph()
graph.add_node("MAIN", my_node)
graph.set_entry_point("MAIN")
graph.add_edge("MAIN", END)

app = graph.compile()

The graph returns the response to the caller immediately. The send_notification coroutine continues running in the background and completes up to 10 seconds later.

Verify it worked

Run the graph and observe the output:

python -m asyncio << 'EOF'
import asyncio
from tenxgraph.core.state import Message
from my_agent import app  # import your compiled graph

result = await app.ainvoke(
    {"messages": [Message.text_message("Start the report")]},
    config={"thread_id": "t1", "user_id": "alice"},
)
print(result["messages"][-1].text())
# Your report is being processed in the background.

await asyncio.sleep(1)  # keep the loop alive so the background task can finish
# Notification sent to alice: Report ready soon
EOF

The key indicator: the graph returns and delivers the response before the background task prints its output.

Set a timeout

Always set a timeout for tasks that do I/O. Without one, a hanging task could leak resources until process shutdown. Timeouts protect against slow or unresponsive external services.

task_manager.create_task(
    upload_to_s3(data),
    name="s3_upload",
    timeout=30.0,          # cancel after 30 seconds
    context={"run_id": config.get("run_id")},  # context for error logs
)

If the task exceeds timeout, it is cancelled and a warning is logged. The cancellation does not affect the graph or the response; it is handled gracefully.

Track task status

Query the task manager to monitor running tasks:

# How many tasks are still running?
count = task_manager.get_task_count()

# Detailed information for all active tasks
for info in task_manager.get_task_info():
    print(f"Task: {info['name']}")
    print(f"  Age: {info['age_seconds']:.1f}s")
    print(f"  Timeout: {info['timeout']}s")
    print(f"  Done: {info['done']}")
    print(f"  Cancelled: {info['cancelled']}")

This is useful for monitoring, debugging, and understanding load on background task execution.

Wait for all tasks before shutdown

If you need to drain the queue before the process exits, call wait_for_all:

await task_manager.wait_for_all(timeout=30.0)

This waits up to 30 seconds for all outstanding tasks to complete. If they do not complete in time, a warning is logged but no exception is raised.

To cancel everything immediately without waiting:

await task_manager.cancel_all()

Use cancel_all when shutting down urgently, or wait_for_all when graceful draining is preferred.

Graceful shutdown integration

The compiled graph shuts the task manager down when it is closed. The shutdown_timeout parameter on compile() controls how long to wait:

app = graph.compile(shutdown_timeout=30.0)

# Later, during process teardown:
await app.aclose()

aclose() calls the task manager’s shutdown(), which cancels all outstanding tasks and then waits up to the timeout for the cancellations to settle. Tasks still running after that are force-cancelled. It does not wait for tasks to finish their work, so if a task must complete, call wait_for_all() first.

Common errors

Related pages

• Stream a graph to understand real-time streaming, an alternative when you need incremental updates.
• Graceful shutdown for production-grade shutdown patterns.
• Dependency injection to learn about other injectable parameters available in nodes.
• Production checklist for shutdown timeout settings in production.