How to Run the Agentic Loop with `needle.agent.run()`: Complete Guide

The Needle.run() method in needle/__init__.py executes a complete → function-call → complete cycle up to max_steps times, automatically dispatching tools and feeding results back to the LLM until the task is finished.

Needle is a lightweight, local-first agent framework from cactus-compute/needle that binds Python callables to a native C-extension inference engine. The run() method is the primary entry point for executing agentic loops—iterative reasoning cycles where an LLM can invoke tools, observe results, and continue reasoning. This guide walks through the method's implementation, configuration options, and practical usage patterns.

Understanding the Agentic Loop Implementation

The core loop lives in needle/__init__.py at lines 39-61. When you invoke agent.run(query), Needle orchestrates a five-stage pipeline:

Stage 1: Initial LLM Completion

The engine calls _complete() to generate a response from your query. The model may return a JSON structure containing "type": "call" and a "function_calls" array describing which tools to invoke.

Stage 2: Tool Dispatch

Each function call is resolved against the internal _functions registry. This registry is populated during initialization via Needle._resolve(), which processes your tools argument. The matching Python callable executes with the provided arguments.

Stage 3: Feedback Completion

Tool results (or errors) are JSON-encoded and injected into the next prompt via _complete(json.dumps(results, ...)). This closes the reasoning loop by giving the LLM observability into tool outputs.

Stage 4: Iteration

Stages 1-3 repeat until either:

• No new function calls are returned (task complete)
• The max_steps limit is reached (safety cutoff)

Stage 5: Result Aggregation

The final response includes a "results" field containing the ordered list of all tool-call outcomes from the session.

Native Engine Binding

The heavy computation is delegated to a native C-extension (libneedle.so|dylib|dll), lazily loaded through helper methods in needle/__init__.py lines 16-54. The _library_path() and _lib() utilities handle platform detection and dynamic loading, ensuring the Python interface remains lightweight.

Setting Up Tools for the Agentic Loop

Needle accepts tools as plain callables, Pydantic models, or decorated functions. Schema generation is handled transparently by needle/agent/tools.py at lines 15-20 via the build_schema implementation.

Basic Tool Definition

from needle.agent.tools import tool, Field

@tool  # automatically registers JSON schema from type hints

def get_weather(city: str, unit: str = "celsius") -> dict:
    """Return a mocked weather report."""
    return {"city": city, "temp": 21, "unit": unit, "summary": "clear"}

The @tool decorator introspects your function signature and docstring to generate the JSON schema required by the LLM for structured function calling.

Pydantic-Based Tool Schema

from pydantic import BaseModel
from needle import Needle

class WeatherRequest(BaseModel):
    city: str
    unit: str = "celsius"

def fetch_weather(req: WeatherRequest) -> dict:
    # simulate external API call

    return {"city": req.city, "temp": 18, "unit": req.unit, "summary": "cloudy"}

agent = Needle(tools=[fetch_weather])

When you pass a callable with Pydantic-typed arguments, Needle._resolve() extracts the model structure and converts it to the expected function-call schema.

Running the Agentic Loop

Basic Usage

from needle import Needle
from needle.agent.tools import tool

@tool
def get_weather(city: str, unit: str = "celsius") -> dict:
    return {"city": city, "temp": 21, "unit": unit, "summary": "clear"}

agent = Needle(
    tools=[get_weather],
    system="You are a helpful assistant that can call tools.",
)

response = agent.run(
    query="What's the weather in Paris today?",
    max_steps=5,           # maximum tool-call cycles

    max_new_tokens=256,    # per-completion token budget

)

print(response)

# Contains final answer and "results" list with tool outputs

Inspecting Tool Results

response = agent.run("Give me the weather in Berlin.")
print(response["results"][0])

# → {"city": "Berlin", "temp": 18, "unit": "celsius", "summary": "cloudy"}

The "results" array preserves the execution order, enabling you to trace the agent's reasoning path through multiple tool invocations.

Key Parameters for agent.run()

Parameter	Type	Purpose
query	str	Initial user prompt that seeds the agentic loop
max_steps	int	Hard limit on complete→call→complete iterations (safety bound)
max_new_tokens	int	Token budget for each LLM completion in the loop

Additional generation parameters are passed through to the underlying native engine via **kwargs in _complete().

Project File Structure

Understanding these source files helps when debugging or extending Needle's agentic capabilities:

• needle/__init__.py — Public API (Needle class), C-extension binding, and run() implementation (source)
• needle/agent/tools.py — Schema generation utilities (build_schema, @tool decorator) (source)
• needle/agent/fetch.py — Platform-specific library discovery (_library_path, _lib_name) (source)
• tests/test_inference.py — Reference implementation showing agent.run() in test scenarios (source)

Summary

• The Needle.run() method implements a complete → function-call → complete cycle for agentic reasoning
• Tool dispatch uses an internal _functions registry populated via Needle._resolve() from your tools argument
• Results are JSON-encoded and fed back through _complete() for iterative refinement
• The native C-extension handles inference, lazily loaded via _library_path() and _lib()
• Schema generation for tools is automatic via needle/agent/tools.py

Frequently Asked Questions

What happens if max_steps is exceeded?

The loop terminates at the step limit and returns the last response, including any "results" collected up to that point. Design your max_steps value based on the expected complexity of multi-tool workflows.

Can I use async functions as tools?

The current implementation in needle/__init__.py expects synchronous callables. The _functions registry stores raw Python callables executed directly in the dispatch phase. For async support, you would need to wrap coroutines with asyncio.run() or similar.

How does Needle handle tool execution errors?

Errors during tool dispatch are caught, serialized, and fed back to the LLM as part of the results payload. This allows the model to observe failures and potentially recover or request alternative approaches in subsequent iterations.

Where is the native library downloaded from?

The needle/agent/fetch.py module implements _library_path() and _lib_name() to locate or download platform-appropriate binaries (libneedle.so, libneedle.dylib, or libneedle.dll). The library is cached locally after first fetch.

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