# How to Execute a Full Agentic Loop with `agent.run()` in Needle

> Learn to execute a full agentic loop with agent.run() in Needle. See how Needle executes Python functions and feeds results back to the model for multi-turn interactions.

- Repository: [Cactus Compute, Inc./needle](https://github.com/cactus-compute/needle)
- Tags: how-to-guide
- Published: 2026-09-01

---

**`agent.run()` drives a complete multi-turn interaction where the model decides which tools to call, Needle executes those Python functions, feeds the results back to the model, and repeats until the model signals completion.**

The Needle library by cactus-compute provides a lightweight, native-backed framework for building tool-using AI agents. At its heart lies the **`agent.run()`** method, which orchestrates the full agentic loop—managing the back-and-forth between language model reasoning and deterministic tool execution without manual intervention.

## How the Agentic Loop Works Internally

Understanding the internal flow of `run()` helps you debug and optimize your agents. The implementation lives in [`needle/__init__.py`](https://github.com/cactus-compute/needle/blob/main/needle/__init__.py) and follows a strict three-phase protocol.

### Phase 1: Initial Completion

When you invoke `run()`, it first calls the private `_complete()` method. This method:
- Initializes the native engine with system facts and tool schemas via `needle_init`
- Sends your user query to the model
- Receives a JSON response containing the first set of `function_calls`

```python

# From needle/__init__.py, lines 39-44

response = self._complete(prompt)  # Returns {"type": "call", "function_calls": [...]}

```

The underlying native library (`libneedle.so`) handles the model's forward pass through `needle_complete`, returning a buffer-encoded JSON payload.

### Phase 2: Tool Execution Loop

For up to `max_steps` iterations (default: 8), `run()` executes:

1. **Validation** – Verifies the response type is `"call"` and pending calls exist
2. **Tool lookup** – Finds each requested function in the internal `_functions` registry
3. **Invocation** – Calls the Python function with supplied arguments, capturing return values or exceptions
4. **Result serialization** – Converts outputs to JSON (handling Pydantic models via `_jsonable`)
5. **Feedback** – Sends serialized results back to the engine with another `_complete()` call

```python

# Conceptual flow from lines 44-60

while steps < max_steps:
    if response["type"] != "call" or not response["function_calls"]:
        break
    
    executed = []
    for call in response["function_calls"]:
        tool = self._functions[call["name"]]
        result = tool(**call["arguments"])
        executed.append(result)
    
    response = self._complete(executed)  # Feed results back to model

    steps += 1

```

### Phase 3: Termination and Result Assembly

The loop terminates when:
- The model returns a non-`"call"` response type (indicating completion or direct answer)
- No `function_calls` are present in the response
- `max_steps` is exhausted

The final response includes a `"results"` field containing the ordered list of all tool executions.

## Basic Usage Examples

### Single-Tool Agent

Start with a simple weather lookup to see the loop in action:

```python
import needle

@needle.tool
def get_weather(city: str):
    """Return a mock weather report."""
    return {"city": city, "temp_c": 22, "sky": "sunny"}

agent = needle.Needle(tools=[get_weather])
result = agent.run("What's the weather in Paris?")
print(result["results"])

# [{'city': 'Paris', 'temp_c': 22, 'sky': 'sunny'}]

```

The model recognizes it needs weather data, emits a `function_call` to `get_weather`, and receives the result—all within one loop iteration.

### Multi-Turn Multi-Tool Workflow

More complex queries require sequential tool use. This example chains search and fetch operations:

```python
import needle

@needle.tool
def search(query: str):
    """Pretend to search the web and return a URL."""
    return {"url": f"https://example.com/search?q={query}"}

@needle.tool
def fetch(url: str):
    """Mock fetch that returns page text."""
    return {"content": f"Fake content from {url}"}

agent = needle.Needle(tools=[search, fetch])
out = agent.run(
    "Find a tutorial on Python decorators and give me the first paragraph",
    max_steps=4
)
print(out["results"])

# [

#   {'url': 'https://example.com/search?q=Python+decorators'},

#   {'content': 'Fake content from https://example.com/search?q=Python+decorators'}

# ]

```

Here the loop executes twice: first `search` returns a URL, then the model decides to `fetch` that URL.

## Advanced Patterns

### Type-Safe Tools with Pydantic

Needle automatically serializes Pydantic models for result transmission. Define return types for clearer contracts:

```python
import needle
from pydantic import BaseModel

class WeatherReport(BaseModel):
    city: str
    temp_c: float
    sky: str

@needle.tool
def get_weather(city: str) -> WeatherReport:
    """Return a typed weather report."""
    return WeatherReport(city=city, temp_c=18.5, sky="cloudy")

agent = needle.Needle(tools=[get_weather])
resp = agent.run("Weather in Berlin?")
print(resp["results"][0].dict())

# {'city': 'Berlin', 'temp_c': 18.5, 'sky': 'cloudy'}

```

The `_jsonable` helper in [`needle/__init__.py`](https://github.com/cactus-compute/needle/blob/main/needle/__init__.py) handles model conversion transparently.

### Tuning Loop Parameters

Control execution boundaries for demanding tasks:

```python
agent.run(
    "Complex task requiring many steps",
    max_steps=12,        # Extend beyond default 8 iterations

    max_new_tokens=512   # Increase per-turn generation budget

)

```

**Important trade-offs:**
- **Higher `max_steps`** enables deeper reasoning chains but increases latency and API costs
- **Higher `max_new_tokens`** allows verbose tool descriptions but may encourage unnecessary verbosity

## Key Source Files and Architecture

| File | Purpose |
|------|---------|
| [`needle/__init__.py`](https://github.com/cactus-compute/needle/blob/main/needle/__init__.py) | Core `Needle` class implementing `run()`, `_complete()`, and native engine binding |
| [`needle/agent/tools.py`](https://github.com/cactus-compute/needle/blob/main/needle/agent/tools.py) | `@needle.tool` decorator and `build_schema` for JSON schema generation |
| [`needle/_telemetry.py`](https://github.com/cactus-compute/needle/blob/main/needle/_telemetry.py) | Usage tracking hooks invoked during loop execution |
| [`tests/test_run.py`](https://github.com/cactus-compute/needle/blob/main/tests/test_run.py) | Integration tests verifying correct multi-turn behavior |

The native engine interface (`needle_init`, `needle_complete`) bridges Python orchestration with optimized model inference, keeping the Python layer thin and inspectable.

## Summary

- **`agent.run()`** abstracts the complete agentic loop: model reasoning → tool execution → result feedback → repeat
- The loop terminates on completion signals, empty call lists, or `max_steps` exhaustion
- Results accumulate in the `"results"` field of the final response
- Use `max_steps` and `max_new_tokens` to balance capability against cost and latency
- Pydantic models integrate seamlessly via automatic JSON serialization

## Frequently Asked Questions

### What's the maximum number of iterations in a single `agent.run()` call?

The default ceiling is **8 iterations**, controlled by the `max_steps` parameter. You can increase this for complex multi-step tasks—set `max_steps=12` or higher—though each additional turn adds latency and token consumption.

### How does Needle handle tool execution errors?

Exceptions during tool invocation are caught, serialized as error objects, and fed back to the model within the `executed` results list. The model receives the error context and can decide whether to retry, use an alternative tool, or report failure to the user.

### Can I inspect intermediate steps during the loop?

Currently, `run()` aggregates all results internally. For step-by-step visibility, you would need to implement a custom loop using the lower-level `_complete()` method, or add logging hooks within your tool functions to emit progress information.

### Does `agent.run()` support parallel tool execution?

The implementation in [`needle/__init__.py`](https://github.com/cactus-compute/needle/blob/main/needle/__init__.py) processes `function_calls` sequentially within each iteration. While the JSON protocol could theoretically support parallel calls, the current Python binding invokes tools one at a time and collects results before the next model turn.