# Beamflow Agents Guide

This guide explains how to build AI Agents and robust integrations using Beamflow. It covers the core concepts you need to know to bring data in, process it, and interact with external systems.

## Core Concepts

Beamflow integrations revolve around the following core ideas:
- **Integration Context**: Automatically tracks the current integration, pipeline, run, and context across distributed operations.
- **Ingress**: How data enters your pipelines (via Webhooks or Polled schedules).
- **Clients**: How you make authenticated requests to external APIs.
- **Records Feed**: How you queue and deduplicate records for processing.
- **Project Structure**: How to organize code between API, Worker, and Shared modules.

## Bringing Data In: Ingress

You can trigger your pipelines through two main ingress decorators provided by `beamflow_lib.pipelines.ingress`:

### 1. Webhooks (`@ingress.webhook`)
Use webhooks when an external system can push events to your application. This decorator registers the handler as part of the integration but leaves the exact handling logic to you.

```python
from beamflow_lib.pipelines.ingress import ingress

@router.post("/webhook/slack")
@ingress.webhook(pipeline="slack.messages", integration="slack")
async def handle_slack_webhook(request):
    data = await request.json()
    # Read the webhook and process the data!
```

### 2. Polling (`@ingress.poll`)
Use polling when you need to fetch data on a schedule. This decorator acts as a specialized scheduler entrypoint that injects a durable `state` dictionary into your function. This is perfect for remembering "watermarks" (like the last processed token or timestamp) to paginate stateful APIs.

The state is automatically saved for you as long as the function executes without exceptions.

```python
from beamflow_lib.pipelines.ingress import ingress
from datetime import datetime

@ingress.poll(pipeline="slack.messages", integration="slack", schedule="*/5 * * * *")
async def poll_slack(state: dict):
    # Retrieve the watermark from the previous run
    since = state.get("since")
    
    # Fetch new data using the watermark...
    page = await fetch_messages_from_api(since=since)
    
    # Process or publish data
    
    # Update the watermark; it will be automatically saved!
    state["since"] = datetime.now()
```

## Making External Requests: Clients

Beamflow makes it easy to interact with external APIs via Clients. They handle setting Base URLs, automatic Authentication headers, observability tracing, and defaults out of the box.

### Adding a Client
To configure a new REST client, you place a configuration file (like `slackClient.yaml`) in your shared clients folder. Beamflow dynamically loads these clients if your configuration registry points to the folder. For example, if you have `_config/shared/clients.yaml` configured as follows:
```yaml
path: clients
pattern: "*.yaml"
```
You can simply define `clients/slackClient.yaml` and the system will expose it to your tasks.

### Using a Client
Once configured, you can retrieve the standard HTTP client (from the `beamflow_clients` package) anywhere.

```python
from beamflow_clients import get_client
from beamflow_lib.decorators import integration_task

# Load the client configuration
slack_client = get_client("slackClient")

@integration_task(integration="slack", integration_pipeline="send_message")
async def send_slack_message_task(message: str):
    # Authorization and base URL logic are automatically handled here
    await slack_client.request("POST", "/chat.postMessage", json={"text": message})
```

### Custom Clients
If you have an API you interact with heavily, you can define a custom typed client. This encapsulates specific API routes for a better developer experience.

```python
from beamflow_lib.clients import HttpClient, client
from typing import List

# Extending HttpClient and registering with the @client decorator
@client("DemoClient")
class DemoClient(HttpClient):
    """Custom client for the Demo API."""
    
    async def get_users(self) -> List[dict]:
        response = await self.request("GET", "/users")
        return response.json()
        
    async def get_user(self, user_id: int) -> dict:
        response = await self.request("GET", f"/users/{user_id}")
        return response.json()

# You can then resolve this custom client by name:
# demo = get_client("DemoClient")
```

## Processing Data: Records Feed

When you receive payloads via Webhooks or Polling, you usually want to process them robustly and asynchronously on the backend. The `RecordsFeed` module provides an opinionated record queue with built-in deduplication that prevents overwhelming pipelines.

### Publishing Records
The framework resolves deduplication using a unique combination of `(integration, record_type, record_id)`. If multiple records arrive with the same identity, "latest-wins" semantics are applied (keeping the one with the latest timestamp).

It also seamlessly supports feeding massive data payloads -- large objects (> 5KB) are automatically offloaded to a BlobStore.

```python
from beamflow_lib.pipelines.records_feed import RecordsFeed
from beamflow_lib.pipelines.records_model import RecordData

# Retrieve your specific feed
feed = RecordsFeed.get(feed_id="slack.messages")

# A common pattern is inserting data retrieved via an `@ingress.poll` into a feed
await feed.publish(RecordData(
    record_id="msg_123",
    record_type="message",
    data={"text": "Hello World", "user": "U123"}
))
```

### Consuming Records
To process the data asynchronously, decorate a function with `@feed_consumer`. This supports robust configurations like automatic batched receiving, delays, rate-limiting, and concurrency control.

```python
from beamflow_lib import feed_consumer, RecordData

@feed_consumer(
    feed_id="slack.messages", 
    batch=True,
    max_batch_size=50,
    max_delay_ms=2000 # wait up to 2 seconds for the queue to fill
)
async def process_batch_messages(records: list[RecordData]):
    """
    Consumes a maximum of 50 records in a batch, waiting 
    up to 2 seconds for the queue to fill.
    """
    print(f"Began processing {len(records)} records.")
    for record in records:
        print(f"Record: {record.record_id} of type {record.record_type}")
        print(f"Data Payload: {record.data}")
```

```python
@feed_consumer(
    feed_id="slack.messages", 
    batch=False
)
async def process_batch_messages(record: RecordData):
    """
    Consumes a single record at a time.
    """
    print(f"Began processing {record.record_id} of type {record.record_type}")
    print(f"Data Payload: {record.data}")
```
## Project Structure

A typical Beamflow project is organized into three main areas: Configuration, Deployment, and Source Code.

```text
.
├── config/                 # Application configuration
│   ├── shared/             # Base configuration for all environments
│   │   ├── backend.yaml    # Task backend settings (shared)
│   │   ├── clients.yaml    # Points to the clients folder
│   │   ├── clients/        # Shared client configurations
│   │   │   └── fooClient.yaml
│   │   └── .env            # Shared environment variables
│   └── local/              # Environment-specific overrides (e.g., local, dev, prod)
│       ├── backend.yaml    # Local-specific backend settings
│       └── .env            # Local-specific environment variables
├── deployment/             # deployment-related files
│   ├── shared/             # Base Dockerfiles
│   │   ├── api.Dockerfile
│   │   └── worker.Dockerfile
│   └── local/              # Local deployment configuration
│       └── docker-compose.yaml
├── src/                    # Source code for your integration
│   ├── api/                # API-specific code (Routes, Webhooks)
│   │   └── routes/
│   │       └── webhooks.py # typical place for @ingress.webhook ... we can also add the webhooks into purpose specific files ie slack_webhooks.py / stripe_webhooks.py etc
│   │       └── routes.py   # we can also add standard api routes ... like health check etc
│   ├── worker/             # Worker-specific code (Tasks, Consumers)
│   │   └── tasks/
│   │       └── xyzTask.py  # place for feed consumers, polling tasks, integration tasks etc
│   └── shared/             # Shared code (Models, Clients, Utils)
│       ├── clients/        # here should to the custom clients
│       └── models/         # here should to the custom models
├── api_main.py             # API Entry point
└── worker_main.py          # Worker Entry point
```

### Source Organization (`src/`)

Beamflow separates code based on where it executes to ensure clean isolation and efficient scaling:

1.  **API (`src/api/`)**: Code that runs on the webserver. Its primary role is to handle incoming requests, typically defined using the `@ingress.webhook` decorator.
2.  **Worker (`src/worker/`)**: Code that runs on the background worker. This is where the "heavy lifting" happens, including `@integration_task`, `@ingress.poll`, and `@feed_consumer` handlers.
3.  **Shared (`src/shared/`)**: Logic used by both the API and Worker, such as custom typed Clients, Pydantic models, and utility functions.

### Configuration & Environments

Beamflow uses a tiered configuration system that allows for seamless transitions between environments:

- **Inheritance**: Configuration is loaded from `config/shared/` first, and then overridden by environment-specific files in `config/{env}/`.
- **Backend Setup**: The `backend.yaml` file defines how tasks are executed (e.g., `asyncio` for local dev or `dramatiq` or `managed` for production).
- **Environment Variables**: `.env` files in both the shared and environment folders are automatically resolved and injected into the application.

### Execution Entry Points

- **`api_main.py`**: The entry point for the web server. It sets up the FastAPI application and automatically imports modules from `src/api` to register routes.
- **`worker_main.py`**: The entry point for the background process. It connects to the task backend (like Redis) and automatically imports modules from `src/worker` to register task signatures.

