How to Execute the Full Agent Loop with `agent.run()` in Needle

Use agent.run() to execute Needle's complete reasoning cycle: the method repeatedly calls the LLM engine, detects function calls, invokes registered Python tools, and feeds results back into the model until it reaches a conclusive answer or hits the step limit.

The Needle framework provides a native agentic loop through the Needle.run() method. This method implements a "think-act-think" pattern that lets large language models reason, call external tools, and refine their answers based on tool outputs. The loop is self-contained in [needle/__init__.py](https://github.com/cactus-compute/needle/blob/main/needle/__init__.py#L83) and requires no additional orchestration code from the user.

How the Agent Loop Works

Needle.run() implements a five-stage reasoning cycle. Understanding these stages helps you debug unexpected behavior and optimize your tool definitions.

Stage 1: Generate Response

The loop begins by calling self._complete() at line 183. This sends the current conversation state to the native Needle engine (or a fine-tuned worker) and receives a JSON envelope containing the model's response.

Stage 2: Detect Function Calls

The method inspects response.get("function_calls") to determine whether the model wants to invoke tools. Function calls are returned as structured JSON objects with tool names and arguments.

Stage 3: Invoke Registered Tools

Valid function calls trigger lookups in self._functions, a dictionary built during the tool-resolution step at needle/__init__.py#L38. The corresponding Python callables execute with the parsed arguments.

Stage 4: Collect and Feed Back Results

Tool return values are JSON-encoded, appended to the execution history, and sent back to the model as user-typed messages. This closes the feedback loop and lets the model reason about the tool outputs.

Stage 5: Repeat Until Completion

The cycle repeats up to max_steps (default 8) or until the model stops issuing function calls. The loop handles missing or unknown tools gracefully—invalid calls are reported back to the model rather than raising exceptions.

Implementing agent.run() in Practice

Basic Single-Tool Execution

This minimal example demonstrates the complete flow with a single echo tool:

from needle import Needle, tool, Field

@tool
def echo(message: str) -> str:
    """Return the same string that was given."""
    return message

agent = Needle(tools=[echo], system="You are a helpful assistant.")
result = agent.run("Please echo the phrase: 'hello world'")
print(result["results"])   # → ['hello world']

The results field contains an ordered list of tool outputs. In this case, only one tool executed, so the list has a single element.

Multi-Step Tool Chaining

More complex queries trigger sequential tool calls. The model may invoke one tool, wait for results, then decide to call another:

from datetime import datetime
from needle import Needle, tool, Field

@tool
def get_time() -> str:
    """Return the current ISO-8601 timestamp."""
    return datetime.utcnow().isoformat()

@tool
def combine(a: str, b: str) -> str:
    """Concatenate two strings with a space."""
    return f"{a} {b}"

agent = Needle(tools=[get_time, combine],
               system="You may call tools to retrieve the time and build a sentence.")
response = agent.run(
    "Create a sentence that says: 'The current time is <time>'."
)
print(response["results"])

Sample output:

['2026-09-05T12:34:56.789012Z',
 'The current time is 2026-09-05T12:34:56.789012Z']

Here the model first calls get_time, receives the timestamp, then invokes combine to construct the final sentence.

Key Implementation Details

Tool Resolution Happens Once

The _resolve() method at needle/__init__.py#L38 converts decorated Python functions into JSON schemas and stores the callable objects in self._functions. This mapping persists for the agent's lifetime.

Engine Abstraction

self._complete() at needle/__init__.py#L61 abstracts the actual inference engine. It routes to either the native Needle engine (downloaded via [needle/agent/fetch.py](https://github.com/cactus-compute/needle/blob/main/needle/agent/fetch.py)) or a fine-tuned worker ([needle/_worker.py](https://github.com/cactus-compute/needle/blob/main/needle/_worker.py)) when a .cact weight file is provided.

Result Aggregation

At needle/__init__.py#L99, the run() method collects all tool outputs into the executed list. When the loop terminates, this list is attached to the final response envelope as the "results" field.

Controlling Loop Behavior

Parameter	Default	Description
max_steps	8	Maximum iterations before forced termination
system	None	System prompt prepended to every conversation
tools	[]	List of @tool-decorated functions available to the model

Pass max_steps explicitly to override the default:

result = agent.run("Complex query requiring many steps", max_steps=12)

Source Files Reference

• needle/__init__.py – Core Needle class with run() implementation and the complete agent loop
• needle/agent/tools.py – @tool decorator and schema builders for Python function exposure
• needle/agent/fetch.py – Native engine library download management
• needle/_worker.py – Fine-tuned worker wrapper for custom .cact models

Summary

• Needle.run() executes the full agent loop: generate response → detect calls → invoke tools → feed back results → repeat
• The method handles up to 8 iterations by default, with graceful handling of invalid tool requests
• Tool outputs are collected in the "results" field of the final response envelope
• All implementation resides in needle/__init__.py, with tool resolution at line 38 and loop control at line 83

Frequently Asked Questions

What happens if the model calls a non-existent tool?

The agent catches unknown tool names during the lookup in self._functions, reports the error back to the model as a failed tool result, and continues with the next iteration. This lets the model recover and attempt a different approach rather than crashing the entire loop.

How does Needle differ from other agent frameworks like LangChain?

Needle uses a native engine approach with direct JSON envelope parsing rather than prompt-based tool formatting. The loop is tightly integrated in needle/__init__.py without intermediate abstraction layers, reducing latency and simplifying debugging for single-container deployments.

Can I use agent.run() with custom fine-tuned models?

Yes. Provide a .cact weight file during Needle initialization to route inference through the fine-tuned worker in needle/_worker.py. The same run() method and tool resolution logic apply regardless of whether you use the base engine or a custom model.

Why does agent.run() return a dictionary instead of a string?

The dictionary format preserves structured metadata including the full message history, token usage, and the ordered results list. Access response["results"] for tool outputs, or response["content"] for the model's final text response if no tools were called.

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