> For the complete documentation index, see [llms.txt](https://fleet.hackmap.win/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://fleet.hackmap.win/guides/inter-automation.md).

# Making two automations talk

This is the headline feature. Once you have more than one automation, you almost always want them connected. Fleet offers four shapes for cross-automation communication:

| Shape              | Primitive          | Use when                                                      |
| ------------------ | ------------------ | ------------------------------------------------------------- |
| One-way pipeline   | `Stream`           | A's output is B's input. Continuous flow, at-most-once.       |
| Discrete dispatch  | `Queue` + `submit` | A enqueues N typed tasks for B's batch automation.            |
| Shared inventory   | `Pool`             | A produces reusable items, many B's borrow them concurrently. |
| Lightweight signal | event bus          | "Something happened" — fire-and-forget broadcast.             |

This page covers all four, with a brief illustration of each. The full CF cookie walkthrough — which uses three of them at once — is in [cf cookies pool](/guides/cf-cookies-pool.md).

## Pipeline (`Stream`)

Producer emits, consumers pop. Destructive, FIFO-ish (oldest first), bounded by length and TTL.

**The wiring rule:** every automation that declares `Output = X` automatically owns a stream **named after its registered name** (the string passed to `@register("…")`). `ctx.emit(payload)` writes to *that* stream. The consumer's `Stream[Ping]("url-pinger", …)` reference must reuse the same name to read it.

```python
# shared contracts package — referenced by both automations
from fleet.core import Stream
from pydantic import BaseModel

class Ping(BaseModel):
    url: str
    ts: float

PINGER_OUT = Stream[Ping]("url-pinger", payload=Ping)
#                          ^^^^^^^^^^^ must match the producer's @register() name
```

```python
# producer
from fleet.core import register

@register("url-pinger")                  # ← this string is the stream name
class UrlPinger(ContinuousAutomation[Cfg]):
    Output = Ping                        # ← declares the producer owns a stream of Pings
    async def run_slot(self, ctx):
        while not ctx.shutdown.is_set():
            await ctx.emit(Ping(url=ctx.config.url, ts=time.time()))
            #     ^^^^^^^^^^^^^^ writes to stream "url-pinger" — framework looks up
            #                    the producer's own_stream() = Stream(automation_type, Output)
            await asyncio.sleep(1)

# consumer — doesn't need its own @register/Output; reads from PINGER_OUT explicitly
@register("logger")
class Logger(ContinuousAutomation[Cfg]):
    async def run_slot(self, ctx):
        while not ctx.shutdown.is_set():
            pings = await ctx.stream(PINGER_OUT).pop(n=10)
            for p in pings:        # p is a Ping instance, type-validated against PINGER_OUT.payload
                ctx.logger.info("pinged %s at %s", p.url, p.ts)
```

The framework's identity: `Stream(automation_type=cls.automation_type, payload=cls.Output)` — see `BaseAutomation.own_stream()` in `src/fleet/core/automation.py`. The producer doesn't pass the stream reference to `emit` because the framework resolves it from the automation's class.

`pop` is the workhorse. It uses `ZPOPMIN` under the hood; two consumers calling `pop` on the same stream don't see duplicates.

Back-pressure: there isn't any. The producer doesn't know whether anyone is consuming. If the producer is faster than all consumers combined, the stream fills up to its length cap and the oldest entries get evicted. For most workloads this is what you want — see [streams and the event bus](/concepts/streams-and-events.md) for the trade-offs.

## Discrete dispatch (`Queue` + `submit`)

For B to be a `BatchAutomation` consuming typed tasks A pushes. The framework validates the payload at submission against B's declared `TaskPayload`.

```python
# shared contracts
from fleet.core import Queue

class ScrapeTask(BaseModel):
    url: str

SCRAPER = Queue[ScrapeTask]("scraper", payload=ScrapeTask)
```

```python
# producer A — could be a ContinuousAutomation or even external code
async def run_slot(self, ctx):
    for url in self._urls():
        await ctx.submit(SCRAPER, ScrapeTask(url=url))

# consumer B — a BatchAutomation
class Scraper(BatchAutomation[Cfg]):
    TaskPayload = ScrapeTask
    Output = ScrapeResult
    async def run_one(self, task, ctx):
        # task.payload is a ScrapeTask
        await fetch(task.payload.url)
```

Submissions can also come from outside the fleet via `POST /api/v1/automations/scraper/tasks` — same schema validation applies. See [batch automation](/guides/batch-automation.md).

## Shared inventory (`Pool`)

For "many borrowers share a small pool of valuable items". The classic case: CF clearance cookies. The harvester produces a few, the scrapers each claim one when they need it, release on exit.

```python
# shared contracts
from fleet.core import Pool

class CfCookie(BaseModel):
    cookies: dict[str, str]
    user_agent: str
    site: str

class CfCookieTags(BaseModel):
    site: str
    fresh: bool = True

COOKIE_POOL = Pool[CfCookie, CfCookieTags]("cf-cookies", payload=CfCookie, tags=CfCookieTags)
```

```python
# producer
class Harvester(BatchAutomation[Cfg]):
    TaskPayload = HarvestRequest
    Pools = {"cookies": COOKIE_POOL}
    async def run_one(self, task, ctx):
        cookies = await harvest_browser(task.payload.site)
        await ctx.pool(COOKIE_POOL).put(
            payload=CfCookie(...), tags=CfCookieTags(site=task.payload.site),
            ttl_seconds=1500,
        )

# consumer
class Scraper(ContinuousAutomation[Cfg]):
    async def run_slot(self, ctx):
        pool = ctx.pool(COOKIE_POOL)
        while not ctx.shutdown.is_set():
            async with pool.where(site=ctx.config.site, fresh=True).claim(hold_seconds=20) as item:
                if item is None:
                    await ctx.submit(HARVESTER, HarvestRequest(site=ctx.config.site))
                    await asyncio.sleep(1)
                    continue
                await use_cookie(item.payload)
```

Pools are non-destructive on read. Multiple consumers can use the same cookie sequentially. A `claim` is an exclusive lease for `hold_seconds` — another slot trying to claim the same item during that window gets nothing.

See [primitives](/concepts/primitives.md) for the full pool API, and [cf cookies pool](/guides/cf-cookies-pool.md) for a complete walkthrough.

## Lightweight signal (event bus)

Fire-and-forget broadcast. No persistence. Use for things you don't mind losing: lifecycle events, debug signals, dashboard updates.

```python
# A:
await ctx.events.publish("harvest.started", {"site": "example.com"})

# B (background task):
async for ev in ctx.events.subscribe("harvest.started"):
    await react_to(ev)
```

If B isn't subscribed at the moment A publishes, the message is gone. That's the correct shape for triggers and the wrong shape for work hand-off. For work, use a stream, a queue, or a pool.

## Cross-host

All four primitives are cross-host by default. The producer and consumer can be on different machines. They communicate through the master's Redis, which both reach via the master's network. There's no direct connection between worker hosts.

For most pipelines this is fine. If you need lower latency or higher throughput than a single Redis can handle, you've outgrown the framework's design point.

## Chaining and topology

You can build any DAG with these primitives. A `cf-cookie-harvester` → `cf-cookies` pool ← many `cf-scrapers` → `cf-scraper` output stream → `cf-stats` aggregator. Each edge is a contract reference declared in a shared package.

The dashboard doesn't currently render the graph (it's on the list). For now, document the topology in your contracts package's README. The graph is implicit in the imports.


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://fleet.hackmap.win/guides/inter-automation.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.