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

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 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

# 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

# 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:

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:

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:

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 handles model conversion transparently.

Tuning Loop Parameters

Control execution boundaries for demanding tasks:

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	Core Needle class implementing run(), _complete(), and native engine binding
needle/agent/tools.py	@needle.tool decorator and build_schema for JSON schema generation
needle/_telemetry.py	Usage tracking hooks invoked during loop execution
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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}