ANTIGRAVITY LABJP
Articles/Agents & Manager
Agents & Manager/2026-04-08Advanced

Antigravity AgentKit 2.0 Runtime Errors: Complete Troubleshooting Guide — tool_call Failures, Infinite Loops, and Context Overflow

A comprehensive guide to diagnosing and fixing AgentKit 2.0 runtime errors in production: tool_call failures, infinite loop detection, context window overflow, parallel agent sync errors, and graceful degradation patterns — all with working code.

Antigravity321AgentKit 2.013Agents18Errors2RuntimeTroubleshooting6Python14

Antigravity AgentKit 2.0 Runtime Errors: Complete Troubleshooting Guide — tool_call Failures, Infinite Loops, and Context Overflow

When you move an AgentKit 2.0 agent from development to production, a class of errors emerges that simply isn't visible during local testing. Tool calls that fail silently, unexpected loops that burn through iterations, context that resets mid-task — almost all of these trace back to either architectural choices or specific implementation pitfalls.

This guide systematically covers the runtime errors unique to AgentKit 2.0, walking through diagnosis and production-ready implementation patterns with full code examples.

Understand the AgentKit 2.0 Execution Model Before Debugging

Knowing the execution model helps you pinpoint failures faster.

from antigravity.agentkit import Agent, Tool, AgentRunner
 
# AgentKit 2.0 basic structure
agent = Agent(
    name="task_agent",
    model="antigravity-pro",
    tools=[tool1, tool2, tool3],
    instructions="You are an agent that executes tasks efficiently.",
    max_iterations=20,  # Required — prevents infinite loops
)
 
runner = AgentRunner(agent=agent)

AgentKit 2.0 agents operate on a repeating cycle: LLM decides → tool executes → result feeds back to LLM → repeat. Errors almost always occur at one of the transitions in this cycle. Knowing which transition is failing is half the battle.

Error Pattern 1: tool_call Failures — Tool Not Found or Wrong Arguments

Symptom: ToolNotFoundError: Tool 'analyze_data' not registered or ToolExecutionError: unexpected argument 'format'.

from antigravity.agentkit import Agent, Tool
from antigravity.agentkit.exceptions import ToolNotFoundError, ToolExecutionError
import logging
 
logger = logging.getLogger(__name__)
 
# ❌ Common mistake: tool name mismatch
def analyze_data_function(data: str, output_format: str = "json") -> dict:
    return {"result": "processed"}
 
tool_wrong = Tool(
    name="analyze",  # Agent calls "analyze_data" → ToolNotFoundError
    function=analyze_data_function
)
 
# ✅ Correct implementation: name and description aligned
def analyze_data(data: str, output_format: str = "json") -> dict:
    """
    Analyze data and return structured results.
 
    Args:
        data: Input to analyze (JSON string or plain text)
        output_format: Output format ("json" or "text")
 
    Returns:
        Dict containing analysis results
    """
    try:
        processed = process_data(data, output_format)
        return {"status": "success", "result": processed}
    except ValueError as e:
        # Return clear error info from inside the tool
        return {"status": "error", "error": str(e), "input_received": data[:100]}
 
tool_correct = Tool(
    name="analyze_data",  # Matches exactly what the agent will call
    function=analyze_data,
    description="Analyzes data. Accepts text or JSON input and returns structured results."
)
 
def build_agent_with_error_handling() -> Agent:
    """Build an agent with proper error handling."""
 
    tools = [tool_correct]
 
    # Confirm tool names at startup
    registered_names = [t.name for t in tools]
    logger.info(f"Registered tools: {registered_names}")
 
    return Agent(
        name="robust_agent",
        model="antigravity-pro",
        tools=tools,
        instructions=f"""
You are a data analysis agent.
Available tools: {', '.join(registered_names)}
If a tool call fails, read the error message carefully, correct the arguments, and retry.
""",
        max_iterations=15
    )

Error Pattern 2: Infinite Loops

Symptom: The agent calls the same tool repeatedly with the same arguments and never reaches a final answer.

from antigravity.agentkit import Agent, AgentRunner
from antigravity.agentkit.callbacks import BaseCallback
from datetime import datetime
from collections import Counter
 
class LoopDetectionCallback(BaseCallback):
    """Callback that detects and interrupts infinite loops."""
 
    def __init__(self, max_same_tool_calls: int = 3, time_window_seconds: int = 60):
        self.tool_call_history = []
        self.max_same_tool_calls = max_same_tool_calls
        self.time_window = time_window_seconds
 
    def on_tool_start(self, tool_name: str, tool_input: dict, **kwargs):
        """Called before each tool invocation."""
        now = datetime.now().timestamp()
 
        self.tool_call_history.append({
            "tool": tool_name,
            "input_hash": hash(str(sorted(tool_input.items()))),
            "timestamp": now
        })
 
        # Expire entries outside the time window
        self.tool_call_history = [
            h for h in self.tool_call_history
            if now - h["timestamp"] < self.time_window
        ]
 
        # Check for repeated identical calls
        recent_calls = [
            h for h in self.tool_call_history
            if h["tool"] == tool_name
            and h["input_hash"] == hash(str(sorted(tool_input.items())))
        ]
 
        if len(recent_calls) >= self.max_same_tool_calls:
            raise RuntimeError(
                f"Infinite loop detected: '{tool_name}' called {len(recent_calls)} times "
                f"with identical arguments. Review the task completion condition."
            )
 
    def on_agent_iteration(self, iteration: int, messages: list, **kwargs):
        """Called at the end of each iteration."""
        if iteration % 5 == 0:
            recent_tools = [h["tool"] for h in self.tool_call_history[-10:]]
            logger.info(f"Iteration {iteration}. Recent tool calls: {Counter(recent_tools)}")
 
# Set clear termination conditions in the agent's instructions
agent = Agent(
    name="loop_safe_agent",
    model="antigravity-pro",
    tools=tools,
    instructions="""
When you have completed the task, always end your final response with:
"TASK_COMPLETE: [brief summary of result]"
 
Complete the goal efficiently without repeating the same operation more than three times.
If you lack information, attempt to fill in the gaps by reasoning before asking the user.
""",
    max_iterations=20,
    callbacks=[LoopDetectionCallback(max_same_tool_calls=3)]
)

Instructions that cause loops (and how to fix them):

# ❌ Vague termination condition — causes loops
bad_instructions = """
Collect information and analyze it.
Use tools as needed.
"""
 
# ✅ Explicit termination condition — prevents loops
good_instructions = """
Complete the task in the following steps:
1. Use search_web to gather information (1–2 calls maximum)
2. Use analyze_data to process the results (1 call maximum)
3. Generate a final answer based on the analysis
 
When all steps are done, end your response with "Done: [summary]".
Even if you feel more data would help, limit data collection to 3 calls total.
"""

Error Pattern 3: Context Window Overflow

Symptom: ContextWindowExceededError, or the agent behaves as though it has forgotten earlier information partway through a long task.

from antigravity.agentkit import Agent
from antigravity.agentkit.memory import SlidingWindowMemory, SummaryMemory
 
# ❌ Default memory settings overflow the context on long tasks
 
# ✅ Sliding window memory: keep only the most recent messages
agent_sliding = Agent(
    name="long_task_agent",
    model="antigravity-pro",
    tools=tools,
    memory=SlidingWindowMemory(
        max_messages=30,      # Retain only the last 30 messages
        preserve_system=True  # Always keep system instructions
    )
)
 
# ✅ Summary memory: compress older context into a rolling summary
agent_summary = Agent(
    name="summarizing_agent",
    model="antigravity-pro",
    tools=tools,
    memory=SummaryMemory(
        summary_interval=10,   # Summarize every 10 messages
        max_summary_length=500 # Max summary length in characters
    )
)
 
# Monitor context usage in real time
class ContextMonitorCallback(BaseCallback):
    def on_llm_response(self, response, **kwargs):
        usage = response.get("usage", {})
        prompt_tokens = usage.get("prompt_tokens", 0)
 
        context_limit = 200_000  # Adjust for your model
        usage_percent = (prompt_tokens / context_limit) * 100
 
        if usage_percent > 70:
            logger.warning(
                f"Context usage: {usage_percent:.1f}% "
                f"({prompt_tokens}/{context_limit} tokens). "
                "Consider compressing memory or splitting the task."
            )
 
# Split large tasks across multiple agents
async def process_large_task(task: str, data_chunks: list) -> dict:
    """Process a large task by splitting it into chunks across independent agents."""
 
    results = []
 
    for i, chunk in enumerate(data_chunks):
        chunk_agent = Agent(
            name=f"chunk_agent_{i}",
            model="antigravity-pro",
            tools=tools,
            max_iterations=10
        )
 
        runner = AgentRunner(agent=chunk_agent)
        result = await runner.run(
            f"Process the following data chunk ({i+1} of {len(data_chunks)}):\n{chunk}"
        )
        results.append(result)
 
    # Aggregate with a dedicated summarization agent
    aggregator = Agent(
        name="aggregator",
        model="antigravity-pro",
        tools=[],  # No tools needed for aggregation
        max_iterations=5
    )
 
    agg_runner = AgentRunner(agent=aggregator)
    return await agg_runner.run(
        f"Combine the following {len(results)} chunk results into a single final answer:\n" +
        "\n---\n".join([str(r) for r in results])
    )

Error Pattern 4: Parallel Agent Synchronization Errors

Symptom: When running multiple agents concurrently, some agents don't receive outputs from others correctly, or execution order is unpredictable.

import asyncio
from antigravity.agentkit import Agent, AgentRunner
from typing import Optional
 
class ParallelAgentOrchestrator:
    """Orchestrates parallel agent execution with timeout and error handling."""
 
    def __init__(self, agents: list[Agent], timeout: float = 120.0):
        self.agents = agents
        self.timeout = timeout
 
    async def run_agent_safely(self, agent: Agent, task: str) -> Optional[dict]:
        """Run a single agent with timeout and error isolation."""
        try:
            runner = AgentRunner(agent=agent)
            result = await asyncio.wait_for(
                runner.run(task),
                timeout=self.timeout
            )
            return {"agent": agent.name, "status": "success", "result": result}
 
        except asyncio.TimeoutError:
            logger.error(f"Agent '{agent.name}' timed out after {self.timeout}s")
            return {"agent": agent.name, "status": "timeout", "result": None}
 
        except Exception as e:
            logger.error(f"Agent '{agent.name}' raised an error: {e}")
            return {"agent": agent.name, "status": "error", "error": str(e), "result": None}
 
    async def run_all(self, tasks: dict[str, str]) -> dict:
        """Run all agents in parallel and report failures individually."""
 
        if len(tasks) != len(self.agents):
            raise ValueError(
                f"Agent count ({len(self.agents)}) doesn't match task count ({len(tasks)})"
            )
 
        coroutines = [
            self.run_agent_safely(agent, tasks[agent.name])
            for agent in self.agents
            if agent.name in tasks
        ]
 
        results = await asyncio.gather(*coroutines, return_exceptions=False)
 
        successful = [r for r in results if r and r["status"] == "success"]
        failed = [r for r in results if r and r["status"] != "success"]
 
        if failed:
            logger.warning(
                f"Parallel run: {len(successful)} succeeded, {len(failed)} failed\n"
                + "\n".join([f"  - {r['agent']}: {r['status']}" for r in failed])
            )
 
        return {
            "successful": successful,
            "failed": failed,
            "total": len(results)
        }
 
# Example usage
async def multi_agent_analysis(document: str) -> dict:
    """Analyze a document using three specialized agents in parallel."""
 
    agents = [
        Agent(name="sentiment_agent", model="antigravity-pro", tools=[sentiment_tool], max_iterations=5),
        Agent(name="summary_agent", model="antigravity-pro", tools=[summary_tool], max_iterations=5),
        Agent(name="keyword_agent", model="antigravity-pro", tools=[keyword_tool], max_iterations=5),
    ]
 
    orchestrator = ParallelAgentOrchestrator(agents=agents, timeout=60.0)
 
    tasks = {
        "sentiment_agent": f"Perform sentiment analysis on the following document:\n{document[:5000]}",
        "summary_agent": f"Summarize the following document in three sentences:\n{document[:5000]}",
        "keyword_agent": f"Extract the 10 most important keywords from the following document:\n{document[:5000]}",
    }
 
    return await orchestrator.run_all(tasks)

Error Pattern 5: Tool Dependency Cycles

Symptom: Tool A requires the output of Tool B, which internally calls Tool A — creating a circular dependency that locks up execution.

from typing import Optional
from functools import wraps
 
class ToolDependencyGraph:
    """Manage tool dependencies and detect cycles before runtime."""
 
    def __init__(self):
        self.dependencies = {}  # tool name -> set of tool names it depends on
 
    def add_dependency(self, tool: str, depends_on: list[str]):
        if tool not in self.dependencies:
            self.dependencies[tool] = set()
        self.dependencies[tool].update(depends_on)
 
    def check_cycles(self) -> list[list[str]]:
        """Detect cycles using depth-first search."""
        visited = set()
        path = []
        cycles = []
 
        def dfs(node):
            if node in path:
                cycle_start = path.index(node)
                cycles.append(path[cycle_start:] + [node])
                return
            if node in visited:
                return
 
            visited.add(node)
            path.append(node)
 
            for dep in self.dependencies.get(node, set()):
                dfs(dep)
 
            path.pop()
 
        for tool in self.dependencies:
            if tool not in visited:
                dfs(tool)
 
        return cycles
 
    def validate(self):
        cycles = self.check_cycles()
        if cycles:
            cycle_str = " → ".join(cycles[0])
            raise ValueError(f"Circular dependency detected in tool graph: {cycle_str}")
        return True
 
# Memoize tool results to prevent redundant calls
def memoize_tool(ttl_seconds: int = 300):
    """Decorator that caches tool function results for the specified TTL."""
    cache = {}
 
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            import time
            cache_key = (args, tuple(sorted(kwargs.items())))
 
            if cache_key in cache:
                result, timestamp = cache[cache_key]
                if time.time() - timestamp < ttl_seconds:
                    logger.debug(f"Cache hit: {func.__name__}({cache_key})")
                    return result
 
            result = func(*args, **kwargs)
            cache[cache_key] = (result, time.time())
            return result
 
        return wrapper
    return decorator
 
@memoize_tool(ttl_seconds=60)
def fetch_user_data(user_id: str) -> dict:
    """Fetch user data, cached for 60 seconds."""
    return api_client.get_user(user_id)

Debugging and Logging AgentKit 2.0 Runtimes

import logging
from antigravity.agentkit.callbacks import BaseCallback
 
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(name)s] %(levelname)s: %(message)s'
)
 
class DetailedExecutionLogger(BaseCallback):
    """Callback that produces structured, detailed logs for AgentKit 2.0 runs."""
 
    def __init__(self, log_level: str = "INFO"):
        self.logger = logging.getLogger("agentkit.execution")
        self.iteration = 0
        self.tool_calls = []
 
    def on_agent_start(self, agent_name: str, task: str, **kwargs):
        self.logger.info(f"=== Agent started: {agent_name} ===")
        self.logger.info(f"Task: {task[:200]}...")
 
    def on_llm_start(self, messages: list, **kwargs):
        self.iteration += 1
        self.logger.debug(f"[Iteration {self.iteration}] LLM call with {len(messages)} messages")
 
    def on_llm_response(self, response, **kwargs):
        content = response.get("content", "")
        tool_calls = response.get("tool_calls", [])
 
        if tool_calls:
            for tc in tool_calls:
                self.logger.info(f"Tool call: {tc['name']}({tc.get('arguments', {})})")
                self.tool_calls.append(tc['name'])
        else:
            self.logger.info(f"Final answer generated: {content[:200]}...")
 
    def on_tool_end(self, tool_name: str, result, **kwargs):
        result_preview = str(result)[:100]
        self.logger.debug(f"Tool result: {tool_name}{result_preview}")
 
    def on_agent_end(self, result, **kwargs):
        from collections import Counter
        tool_summary = Counter(self.tool_calls)
        self.logger.info("=== Agent completed ===")
        self.logger.info(f"Total iterations: {self.iteration}")
        self.logger.info(f"Tool call counts: {dict(tool_summary)}")
 
# Graceful degradation with a fallback agent
class AgentWithFallback:
    """Wraps an agent with automatic fallback on repeated failure."""
 
    def __init__(self, primary_agent: Agent, fallback_agent: Agent):
        self.primary = primary_agent
        self.fallback = fallback_agent
 
    async def run(self, task: str, max_retries: int = 2) -> dict:
        for attempt in range(max_retries):
            try:
                runner = AgentRunner(
                    agent=self.primary,
                    callbacks=[DetailedExecutionLogger()]
                )
                result = await runner.run(task)
                return {"agent": "primary", "result": result}
 
            except RuntimeError as e:
                if "infinite loop" in str(e).lower() or "max_iterations" in str(e):
                    logger.warning(f"Primary agent failed (attempt {attempt + 1}): {e}")
                    if attempt == max_retries - 1:
                        logger.info("Switching to fallback agent")
                        fallback_runner = AgentRunner(agent=self.fallback)
                        result = await fallback_runner.run(task)
                        return {"agent": "fallback", "result": result}
                else:
                    raise
 
        raise RuntimeError(f"All {max_retries} attempts failed")

Looking back

The main AgentKit 2.0 runtime error patterns and their fixes:

  • tool_call failures → Verify tool name alignment, validate argument types, return clear error info from inside tool functions
  • Infinite loops → Set max_iterations, use LoopDetectionCallback, write instructions with explicit termination conditions
  • Context overflow → Use SlidingWindowMemory or SummaryMemory, split large tasks with an aggregation agent
  • Parallel sync errors → Use timeout-guarded parallel execution, record failures per agent, design for partial success
  • Tool dependency cycles → Pre-validate with ToolDependencyGraph, cache repeated calls with memoize_tool

Before deploying to production, run DetailedExecutionLogger on real workloads to visualize execution patterns. Catching unexpected loops or excessive tool calls at this stage is far less costly than debugging them in production.

For a deeper look at production multi-agent orchestration patterns, see Antigravity Multi-Agent Production Orchestration Guide.

Share

Thank You for Reading

Antigravity Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.

  • Copy-paste ready implementation code
  • New advanced guides published daily
  • $5/mo or $10 for lifetime access
View Membership →

If you found this article helpful, a small tip ($1.50) would mean a lot to us. Your support helps keep this site ad-free and covers server and hosting costs.

Related Articles

Agents & Manager2026-06-14
Making My Managed Agents Batch Survive a Crash Without Redoing Everything
Running a 200-item batch on the Managed Agents API kept torching tokens, because every mid-run failure restarted from item one. Here is the checkpoint-and-idempotency design I added so the batch resumes from where it died.
Agents & Manager2026-06-12
Running Gemini's Managed Agents API: Where Cloud Execution Ends and My Local Agents Begin
A hands-on record of launching Gemini's Managed Agents (public preview) from Python — polling, artifact retrieval, and a cost guard — plus five criteria I use to decide what stays on my local CLI agents.
Agents & Manager2026-07-10
Tracing Why an Agent Wrote That Line Six Months Ago — Commit Granularity and Provenance Trailers
When an agent packs 14 files and 800 lines into a single commit, git blame tells you nothing six months later. Here is how I split commits at intent boundaries, recorded provenance as machine-readable Git trailers, and built a one-command path from a blamed line back to the design decision behind it.
📚RECOMMENDED BOOKS
Build a Large Language Model (From Scratch)
Sebastian Raschka
LLM Dev
Prompt Engineering for LLMs
Berryman & Ziegler
Prompting
AI Engineering
Chip Huyen
AI Eng
* Contains affiliate links
See all →