How to Manually Control the Agent Loop Using `agent.complete()` in Needle 2

Use agent.complete() in Needle 2 to execute a single turn of the agent loop, inspect the model's proposed function calls, execute them yourself, and feed results back for subsequent turns.

Needle 2 by cactus-compute provides two distinct ways to drive agentic interactions. While agent.run() handles the complete loop automatically, agent.complete() exposes the underlying mechanics for developers who need fine-grained control over tool execution, logging, or external system integration. This guide explains the manual loop pattern with precise implementation details from the Needle source code.

agent.run() vs agent.complete(): Understanding the Architecture

Needle 2 implements two complementary methods in needle/__init__.py that serve different use cases:

• agent.run(query, …) — Executes the full agentic loop internally. The model decides which tool to call, Needle runs the Python function, feeds the result back, and repeats until completion. This happens transparently without caller intervention.

• agent.complete(text, …) — Performs exactly one turn. It sends a prompt to the model and returns the raw response, including any suggested function calls. You must handle execution and feedback yourself.

The manual approach mirrors the internal logic of run() (see lines 40-60 of needle/__init__.py) but inserts a hook between every model request and tool execution.

When to Use Manual Loop Control

Consider driving the loop manually when you need to:

• Interleave custom logic between tool calls (validation, rate limiting, side effects)
• Log or audit intermediate results before the model sees them
• Integrate with external systems that require synchronous handoff
• Implement custom retry or fallback logic for failed tool calls
• Stream progress to users rather than batching the full response

The Five-Step Manual Loop Pattern

According to the API documentation in doc/apis.md (line 70) and the implementation in needle/__init__.py (line 119), controlling the loop manually follows this protocol:

1. Call complete with the initial user query
2. Inspect the response — check if response["type"] == "call" for pending function calls
3. Execute each call using the registered Python functions from your tools parameter
4. Feed results back by calling complete() again with json.dumps() of the results
5. Repeat until response["type"] is not "call" (or function_calls is empty)

Complete Working Example

This implementation demonstrates the full manual loop with a practical smart home scenario:

import json
import needle

# Step 1: Declare a simple tool with the @needle.tool decorator

@needle.tool
def set_lights(room: str, on: bool, brightness: int | None = None):
    """Control a room's lights."""
    return {"room": room, "on": on, "brightness": brightness}

# Step 2: Create the agent with the tool set

agent = needle.Needle(tools=[set_lights])

# Step 3: First turn — ask the model to act

resp = agent.complete("dim the living room to 30")
print(resp)  # => contains "type": "call" with function_calls array

# Step 4: Execute the suggested call(s)

results = []
for call in resp.get("function_calls", []):
    fn = agent._functions[call["name"]]
    results.append(fn(**call["arguments"]))

# Step 5: Feed the execution result back to the model

# The model expects a JSON string representing the result

next_resp = agent.complete(json.dumps(results))
print(next_resp)  # May produce another call or final answer

# Step 6: Loop until the model stops calling tools

while next_resp.get("type") == "call":
    results = []
    for call in next_resp.get("function_calls", []):
        fn = agent._functions[call["name"]]
        results.append(fn(**call["arguments"]))
    next_resp = agent.complete(json.dumps(results))

# Step 7: Final response contains accumulated tool results

print("Final results:", next_resp.get("results"))

Key Implementation Details

Response Format from complete()

The complete() method returns a JSON envelope with a consistent structure documented in doc/apis.md (line 84):

{
  "type": "call",
  "function_calls": [
    {
      "name": "set_lights",
      "arguments": {"room": "living room", "on": true, "brightness": 30}
    }
  ]
}

A final response uses "type": "complete" and contains accumulated results in the "results" field.

Tool Registration and Schema Generation

Tools are registered via the @needle.tool decorator defined in needle/agent/tools.py. This decorator invokes build_schema() to generate a JSON schema that the model can invoke. The same functions passed to needle.Needle(tools=[...]) become accessible via agent._functions for manual execution.

Critical Serialization Requirement

After executing tool calls, you must serialize results with json.dumps() before passing them back to complete(). The engine parses this string and continues the conversation. Passing raw Python objects will cause parsing errors.

Production-Ready Loop with Error Handling

For robust implementations, add error handling and logging:

import json
import needle
from typing import Any

def execute_manual_loop(agent: needle.Needle, query: str) -> dict[str, Any]:
    """Execute agent loop with full observability."""
    turn = 0
    resp = agent.complete(query)
    
    while resp.get("type") == "call":
        turn += 1
        print(f"Turn {turn}: {len(resp.get('function_calls', []))} call(s)")
        
        results = []
        for call in resp.get("function_calls", []):
            try:
                fn = agent._functions[call["name"]]
                result = fn(**call["arguments"])
                results.append({
                    "name": call["name"],
                    "result": result,
                    "status": "success"
                })
            except Exception as e:
                results.append({
                    "name": call["name"],
                    "error": str(e),
                    "status": "error"
                })
        
        resp = agent.complete(json.dumps(results))
    
    print(f"Completed in {turn} turn(s)")
    return resp

Summary

• agent.complete() executes one turn of the agent loop, returning raw model responses with proposed function calls
• Manual control requires you to execute calls via agent._functions and feed json.dumps() results back
• Loop termination occurs when response["type"] != "call" or function_calls is empty
• Source implementation resides in needle/__init__.py (lines 40-60 for run(), line 119 for complete())
• Tool schemas are generated by needle/agent/tools.py and bound to agent._functions at initialization

Frequently Asked Questions

How does agent.complete() differ from agent.run() internally?

Both methods use the same underlying model interface in needle/__init__.py. The run() method wraps complete() in a loop (lines 40-60) that automatically executes calls and feeds results back. When you use complete() directly, you take responsibility for this loop implementation.

Can I mix run() and complete() in the same application?

Yes. You can use run() for straightforward interactions and drop down to complete() when specific turns require custom handling. Maintain separate Needle instances if you need different tool sets for each pattern.

What happens if I pass malformed JSON back to complete()?

The model will likely fail to parse the result or generate an error response. Always use json.dumps() for serialization. For complex objects, ensure they are JSON-serializable or convert them to dictionaries first.

Is there a maximum number of turns for manual loops?

No hard limit exists in complete() itself. Your loop termination logic determines when to stop. For safety, implement a maximum iteration counter in production code to prevent infinite loops from ambiguous model responses.

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