# Calling and Fan-Out
source: https://docs.chalk.ai/docs/functions/calling-and-fan-out

## Call deployed Chalk Functions with defer() and get(), fan out with gather(), compose nested calls, and handle errors.

This page covers deferred calls, fan-out, and nested calls to a deployed Chalk Function, and how its errors reach the caller.

### Deferred calls with defer() and get()

.remote(...) blocks for the entire round trip. Use
.defer(...) instead when you want to submit a call and collect its result
later; it enqueues the call and returns a FunctionCallHandle immediately,
without waiting for the function to run. This example calls the add function from
Defining a function:

```
import chalkcompute

add = chalkcompute.RemoteFunction.from_name("add")

handle = add.defer(3, 5)
# ...do other work while the call runs...
result = handle.get()
print(result)  # 8
```

handle.get(timeout=None) blocks until the result is ready and returns the
same value .remote(...) would have returned.

.defer(...) uses the function queue. It works on a
function version deployed after the function queue's dataplane Redis was enabled. A version
deployed before then accepts .remote(...) calls only.

Each handle exposes the server-assigned call_id it was enqueued under:

```
handle = add.defer(3, 5)
print(handle.call_id)
```

### Fan-out and gather()

Because .defer(...) returns immediately, you can enqueue many calls up
front and collect their results afterward. Use chalkcompute.gather(...) to
wait for a batch of handles concurrently and get their results back in order:

```
import chalkcompute

handles = [classify.defer(text) for text in texts]
results = chalkcompute.gather(handles)
```

gather(...) waits for the handles concurrently rather than one at a time,
so it returns as soon as the slowest call finishes. Pass timeout to bound the whole wait:

```
results = chalkcompute.gather(handles, timeout=30.0)
```

Inside a running event loop, gather(...) returns an awaitable instead of
blocking: await chalkcompute.gather(handles).

Each deferred call is still subject to the function's own
concurrency and rate limit
policies, so fanning out this way doesn't bypass those caps.

### Timeouts

Pass timeout (seconds) to bound how long get() waits:

```
try:
    result = handle.get(timeout=5.0)
except TimeoutError:
    print("still running")
```

A timeout only stops waiting locally. It does not cancel the underlying
call, which keeps running and can still be collected with another get().

### Errors from deferred calls

If the function raised an exception, .get() raises
chalkcompute.RemoteExecutionError, as described in Error handling. Transient
infrastructure failures, such as a dropped connection between the queue and
the worker, are retried automatically before get() ever sees them; only
failures from the function body itself, or infra failures that exhaust
retries, are raised to the caller.

### Results from generator functions

Calling .remote(...) on an async def generator function returns an async iterator of pyarrow
RecordBatch objects. When the generator yields single values, such as strings or numbers, each
batch has one column, result, and batch.column("result").to_pylist() returns the values. When it
yields a TypedDict, each key is a column, and batch.to_pylist() returns the yielded dictionaries;
see Streaming results.

### Nested function calls

A function can call other functions with .remote(...). Each inner call is
dispatched independently, so you can compose pipelines where each step runs on
its own infrastructure. In this example, sentence_stats calls the split_sentences
generator and the count_words function:

```
from collections.abc import AsyncIterator

import chalkcompute


@chalkcompute.function
async def split_sentences(text: str) -> AsyncIterator[str]:
    for sentence in text.split(". "):
        yield sentence.rstrip(".")


@chalkcompute.function
def count_words(sentence: str) -> int:
    return len(sentence.split())


@chalkcompute.function
async def sentence_stats(text: str) -> AsyncIterator[str]:
    async for batch in split_sentences.remote(text):
        for sentence in batch.column("result").to_pylist():
            words = await count_words.remote(sentence)
            yield f"{sentence}: {words} words"
```

The caller sees a single stream of results:

```
import asyncio

async def main():
    text = "Chalk runs each function on its own workers. Calls go through a queue."
    async for batch in sentence_stats.remote(text):
        for line in batch.column("result").to_pylist():
            print(line)

if __name__ == "__main__":
    asyncio.run(main())
```

```
Chalk runs each function on its own workers: 8 words
Calls go through a queue: 5 words
```

The outer function composes two inner calls: split_sentences (async generator)
and count_words (single-value). Each runs on its own compute configuration.

### Error handling

If a function raises an exception, the caller receives a chalkcompute.RemoteExecutionError. Its
exception_type and message attributes hold the original exception's type name and message, and
printing the error shows the remote traceback. RemoteExecutionError is a subclass of
chalkcompute.FunctionError, which the SDK also raises for failures outside your function, such as
expired call results.

```
import chalkcompute

@chalkcompute.function
def divide(a: float, b: float) -> float:
    return a / b

if __name__ == "__main__":
    try:
        divide.remote(1.0, 0.0)
    except chalkcompute.RemoteExecutionError as e:
        print(f"{e.exception_type}: {e.message}")  # ZeroDivisionError: division by zero
```

For async generators, an exception mid-stream terminates the iterator. Any values
already yielded are still available to the caller; only subsequent iteration raises:

```
import asyncio
from collections.abc import AsyncIterator

import chalkcompute

@chalkcompute.function
async def fragile_gen() -> AsyncIterator[str]:
    yield "ok-1"
    yield "ok-2"
    raise ValueError("something went wrong")

async def main():
    try:
        async for batch in fragile_gen.remote():
            for item in batch.column("result").to_pylist():
                print(item)
    except chalkcompute.RemoteExecutionError as e:
        print(f"{e.exception_type}: {e.message}")

if __name__ == "__main__":
    asyncio.run(main())
# ok-1
# ok-2
# ValueError: something went wrong
```