Skip to content

Functional API

The functional API is the Tulip's "agent as a task" shape — @task and @entrypoint decorators that bring agent runs into the regular asyncio universe.

Functional pattern — @entrypoint at top fans out to multiple @task agents via asyncio.gather, results merge into a list

What it is

Two decorators:

Decorator What it does
@task Wraps a coroutine that calls an Agent. Returns a Task you can await, gather, retry, time-out — anything asyncio gives you.
@entrypoint Marks the top-level coroutine of a workflow. You await it like any coroutine; it also records the last run on .last_result / .get_result() for inspection.

These are not a new orchestration runtime. They're a thin shim that lets agents participate in plain asyncio. The point is to compose with asyncio.gather, asyncio.wait_for, asyncio.Queue, or anything else you already use.

When to use it

  • ✅ You think in async def and asyncio.gather already.
  • ✅ The flow is map/reduce over agents (enrich N IOCs in parallel).
  • ✅ You want to mix agents with non-agent code — DB writes, HTTP calls, file I/O — in the same coroutine.
  • ✅ Tooling like tenacity retries, asyncio.timeout, or a asyncio.Queue scheduler already gives you the orchestration you need.

When NOT to use it

  • ❌ You want inspectable, named control-flow with cycles or conditional branches → use StateGraph.
  • ❌ You need per-node retry / cache policies as data → StateGraph.
  • ❌ Different agents should decide who runs → use Orchestrator.

Code

import asyncio
from tulip.multiagent.functional import task, entrypoint

@task
async def enrich_ioc(ioc: dict) -> dict:
    """Run the triage agent against one indicator."""
    return triage_agent.run_sync(f"Enrich {ioc['value']}.").message

@entrypoint
async def enrich_all(iocs: list[dict]) -> list[dict]:
    """Enrich every indicator in parallel; gather the results."""
    return await asyncio.gather(*[enrich_ioc(i) for i in iocs])

scored = await enrich_all(indicators)        # or: asyncio.run(enrich_all(indicators))

Map/reduce with retries and timeouts

Because tasks are plain coroutines, you compose with whatever the asyncio ecosystem provides:

from tenacity import retry, stop_after_attempt, wait_exponential

@task
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=0.5))
async def enrich_ioc(ioc: dict) -> dict:
    return triage_agent.run_sync(f"Enrich {ioc['value']}.").message

@entrypoint
async def enrich_all_with_deadline(iocs: list[dict]) -> list[dict]:
    async with asyncio.timeout(60):                # 60s wall-clock cap
        return await asyncio.gather(*[enrich_ioc(i) for i in iocs])

Tasks calling tasks

Tasks compose. An @entrypoint workflow can call other @tasks including parallel batches inside sequential phases:

@task
async def prioritize_alerts(queue: list[dict]) -> list[dict]:
    return triage_agent.run_sync(f"Pick the top 5 from {len(queue)}.").message

@task
async def enrich(alert: dict) -> dict:
    return triage_agent.run_sync(f"Enrich {alert['ioc']}.").message

@entrypoint
async def end_to_end(queue: list[dict]) -> dict:
    shortlisted = await prioritize_alerts(queue)            # phase 1
    scored = await asyncio.gather(*[enrich(a) for a in shortlisted])  # phase 2 (parallel)
    final = containment_agent.run_sync(f"Recommend containment from: {scored}").message  # phase 3
    return final

Notebooks

Source

multiagent/functional.py — task, entrypoint, TaskResult, EntrypointResult.

See also