Antigravity × LangGraph: Complete Masterguide for Stateful AI Agent Development

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
 
# Define agent state
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]  # Message history (append mode)
    current_step: str          # Name of current processing step
    tool_results: dict         # Tool execution results
    iteration_count: int       # Loop counter (prevents infinite loops)
    error: str | None          # Error information
 
# Initialize graph
graph = StateGraph(AgentState)

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
 
# Initialize LLM
llm = ChatOpenAI(model="gemini-2.0-flash", temperature=0)
 
# Define tools
@tool
def search_codebase(query: str) -> str:
    """Search the codebase and return relevant code snippets"""
    # In practice, this integrates with Antigravity's MCP tools
    return f"Search results for: {query}"
 
@tool
def run_tests(test_path: str) -> dict:
    """Run tests at the specified path and return results"""
    return {"passed": 10, "failed": 2, "errors": ["test_auth.py:45"]}
 
tools = [search_codebase, run_tests]
llm_with_tools = llm.bind_tools(tools)
 
# Agent node (LLM invocation)
def agent_node(state: AgentState) -> dict:
    """Main LLM agent processing step"""
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "current_step": "agent",
        "iteration_count": state.get("iteration_count", 0) + 1
    }
 
# Tool execution node
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)
 
# Conditional edge routing function
def should_continue(state: AgentState) -> str:
    """Determine the next node based on current state"""
    messages = state["messages"]
    last_message = messages[-1]
 
    # Max iteration check (prevents infinite loops)
    if state.get("iteration_count", 0) >= 10:
        return "end"
 
    # Route to tools if there are tool calls
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
 
    # Otherwise, finish
    return "end"
 
# Build the graph
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
 
graph.set_entry_point("agent")
graph.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", "end": END}
)
graph.add_edge("tools", "agent")  # Always return to agent after tool execution
 
app = graph.compile()

# pyproject.toml (Antigravity-generated example)
[project]
name = "langgraph-agent"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "langgraph>=0.2.0",
    "langchain>=0.3.0",
    "langchain-google-genai>=2.0.0",
    "langchain-community>=0.3.0",
    "psycopg[binary]>=3.1.0",
    "langgraph-checkpoint-postgres>=2.0.0",
    "python-dotenv>=1.0.0",
    "pydantic>=2.0.0",
]
 
[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pytest-asyncio>=0.23.0",
    "pytest-mock>=3.12.0",
    "ruff>=0.3.0",
]

import asyncio
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
 
async def create_agent_with_checkpoints():
    """Create an agent with PostgreSQL checkpointing"""
 
    connection_string = "postgresql://user:password@localhost:5432/langgraph_db"
 
    async with AsyncConnectionPool(
        conninfo=connection_string,
        max_size=20,
        kwargs={"autocommit": True, "prepare_threshold": 0}
    ) as pool:
        # Initialize checkpointer
        checkpointer = AsyncPostgresSaver(pool)
        await checkpointer.setup()  # Auto-creates required tables
 
        # Compile graph with checkpointer
        app = graph.compile(checkpointer=checkpointer)
 
        # Use thread_id to maintain conversation continuity
        config = {
            "configurable": {
                "thread_id": "user-123-task-456",
                "checkpoint_ns": "code_review_agent"
            }
        }
 
        # First invocation
        initial_state = {
            "messages": [HumanMessage(content="Please review this PR")],
            "current_step": "start",
            "tool_results": {},
            "iteration_count": 0,
            "error": None
        }
 
        result = await app.ainvoke(initial_state, config=config)
        print(f"First run result: {result['messages'][-1].content}")
 
        # Resume later using the same thread_id (continues from saved state)
        followup = {
            "messages": [HumanMessage(content="Focus especially on the security section")]
        }
        result2 = await app.ainvoke(followup, config=config)
 
        return result2

async def inspect_checkpoints(app, thread_id: str):
    """Inspect the checkpoint history for a given thread"""
    config = {"configurable": {"thread_id": thread_id}}
 
    # Get current state
    current_state = await app.aget_state(config)
    print("Current state values:", current_state.values)
    print("Next nodes:", current_state.next)
    print("Created at:", current_state.created_at)
 
    # Retrieve checkpoint history (most recent 3)
    history = []
    async for state in app.aget_state_history(config, limit=3):
        history.append({
            "checkpoint_id": state.config["configurable"]["checkpoint_id"],
            "step": state.values.get("current_step"),
            "message_count": len(state.values.get("messages", [])),
            "created_at": state.created_at
        })
 
    return history
 
# Roll back to a specific checkpoint
async def rollback_to_checkpoint(app, thread_id: str, checkpoint_id: str):
    """Roll back to a specific checkpoint and re-run from that point"""
    config = {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_id": checkpoint_id  # Target checkpoint to restore
        }
    }
    result = await app.ainvoke(
        {"messages": [HumanMessage(content="Try a different approach")]},
        config=config
    )
    return result

from langgraph.graph import StateGraph, END
 
# High-risk node requiring human approval
def deploy_to_production(state: AgentState) -> dict:
    """Deploy to production (requires approval via interrupt_before)"""
    deploy_result = execute_deployment(state["tool_results"]["build_artifact"])
    return {
        "messages": [AIMessage(content=f"Deployment complete: {deploy_result}")],
        "current_step": "deployed"
    }
 
def prepare_deployment(state: AgentState) -> dict:
    """Prepare deployment (build, test, etc.)"""
    return {
        "tool_results": {"build_artifact": "dist/app.zip", "test_passed": True},
        "current_step": "deploy_ready",
        "messages": [AIMessage(content="Build and tests passed. Ready to deploy to production — please confirm.")]
    }
 
# Build deployment graph
deploy_graph = StateGraph(AgentState)
deploy_graph.add_node("prepare", prepare_deployment)
deploy_graph.add_node("deploy", deploy_to_production)
deploy_graph.set_entry_point("prepare")
deploy_graph.add_edge("prepare", "deploy")
deploy_graph.add_edge("deploy", END)
 
# Pause before the "deploy" node for human review
deploy_app = deploy_graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["deploy"]
)
 
# Execute — agent will pause before "deploy"
config = {"configurable": {"thread_id": "deploy-task-001"}}
state = await deploy_app.ainvoke(initial_state, config=config)
 
# At this point, the agent is paused before the deploy node
current = await deploy_app.aget_state(config)
print("Waiting for approval:", current.next)  # ['deploy']
print("Build result:", current.values["tool_results"])
 
# Human approves — pass None to continue without changing state
await deploy_app.ainvoke(None, config=config)
 
# Human approves with additional instructions
await deploy_app.aupdate_state(
    config,
    {"messages": [HumanMessage(content="Approved. Please also deploy to staging simultaneously.")]}
)
await deploy_app.ainvoke(None, config=config)

from langgraph.graph import StateGraph, END
from typing import Literal
 
class SupervisorState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    next_agent: str
    task_context: dict
    completed_tasks: List[str]
 
# Supervisor node
def supervisor_node(state: SupervisorState) -> dict:
    """Decide which specialized agent to run next"""
 
    system_prompt = """You are an agent orchestrator.
    Coordinate the following specialized agents to complete tasks:
    - researcher: information gathering and research
    - coder: code implementation and fixes
    - reviewer: code review and testing
    - documenter: documentation generation
    - FINISH: all tasks complete
 
    Respond with exactly one word: the name of the next agent to run.
    """
 
    response = llm.invoke([
        {"role": "system", "content": system_prompt},
        *state["messages"]
    ])
 
    return {"next_agent": response.content.strip()}
 
# Specialized agent (researcher example)
def researcher_agent(state: SupervisorState) -> dict:
    """Specialized research agent"""
    research_llm = llm_with_tools  # LLM equipped with web_search, read_doc tools
 
    result = research_llm.invoke([
        {"role": "system", "content": "You are a technical research specialist."},
        *state["messages"]
    ])
 
    return {
        "messages": [result],
        "completed_tasks": state.get("completed_tasks", []) + ["research"]
    }
 
# Routing function
def route_to_agent(state: SupervisorState) -> Literal["researcher", "coder", "reviewer", "documenter", "end"]:
    agent = state["next_agent"].lower()
    if agent == "finish":
        return "end"
    return agent if agent in ["researcher", "coder", "reviewer", "documenter"] else "end"
 
# Build supervisor graph
supervisor_graph = StateGraph(SupervisorState)
supervisor_graph.add_node("supervisor", supervisor_node)
supervisor_graph.add_node("researcher", researcher_agent)
supervisor_graph.add_node("coder", coder_agent)
supervisor_graph.add_node("reviewer", reviewer_agent)
supervisor_graph.add_node("documenter", documenter_agent)
 
supervisor_graph.set_entry_point("supervisor")
supervisor_graph.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "researcher": "researcher",
        "coder": "coder",
        "reviewer": "reviewer",
        "documenter": "documenter",
        "end": END
    }
)
 
# All specialized agents return to supervisor after completion
for agent in ["researcher", "coder", "reviewer", "documenter"]:
    supervisor_graph.add_edge(agent, "supervisor")
 
multi_agent_app = supervisor_graph.compile(checkpointer=checkpointer)

import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
 
def agent_node_with_retry(state: AgentState) -> dict:
    """Agent node with retry logic built in"""
 
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((Exception,)),
        reraise=True
    )
    def _invoke_with_retry():
        return llm_with_tools.invoke(state["messages"])
 
    try:
        response = _invoke_with_retry()
        return {
            "messages": [response],
            "current_step": "agent",
            "error": None,
            "iteration_count": state.get("iteration_count", 0) + 1
        }
    except Exception as e:
        error_msg = f"Agent execution error: {str(e)}"
        return {
            "messages": [AIMessage(content=f"An error occurred: {error_msg}")],
            "current_step": "error",
            "error": error_msg
        }
 
# Error recovery node
def error_handler_node(state: AgentState) -> dict:
    """Handle error state and attempt recovery"""
    error = state.get("error", "Unknown error")
 
    recovery_response = llm.invoke([
        *state["messages"],
        HumanMessage(content=f"An error occurred: {error}\nPlease try an alternative approach to achieve the goal.")
    ])
 
    return {
        "messages": [recovery_response],
        "current_step": "recovery",
        "error": None  # Clear the error
    }
 
# Route on error
def route_on_error(state: AgentState) -> str:
    if state.get("error"):
        return "error_handler"
    return should_continue(state)
 
graph.add_node("error_handler", error_handler_node)
graph.add_conditional_edges("agent", route_on_error, {
    "tools": "tools",
    "end": END,
    "error_handler": "error_handler"
})
graph.add_edge("error_handler", "agent")  # Retry after recovery

import os
from langsmith.run_helpers import traceable
 
# Enable LangSmith via environment variables
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "YOUR_LANGSMITH_API_KEY"
os.environ["LANGCHAIN_PROJECT"] = "antigravitylab-agent"
 
@traceable(name="code_review_workflow", metadata={"version": "1.0", "env": "production"})
async def run_code_review_agent(pr_url: str, thread_id: str) -> dict:
    """Run code review agent with full LangSmith tracing"""
    config = {"configurable": {"thread_id": thread_id}}
 
    result = await app.ainvoke(
        {"messages": [HumanMessage(content=f"Please review this PR: {pr_url}")]},
        config=config
    )
 
    return {
        "review": result["messages"][-1].content,
        "steps_taken": result.get("iteration_count", 0)
    }

import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from langchain_core.messages import HumanMessage, AIMessage
 
@pytest.mark.asyncio
async def test_agent_node_normal_flow():
    """Test normal execution path of agent node"""
    mock_response = AIMessage(
        content="I'll search the codebase",
        tool_calls=[{
            "name": "search_codebase",
            "args": {"query": "authentication"},
            "id": "call_001"
        }]
    )
 
    with patch.object(llm_with_tools, "invoke", return_value=mock_response):
        state = AgentState(
            messages=[HumanMessage(content="Look into the auth code")],
            current_step="start",
            tool_results={},
            iteration_count=0,
            error=None
        )
 
        result = agent_node(state)
 
        assert result["current_step"] == "agent"
        assert result["iteration_count"] == 1
        assert len(result["messages"]) == 1
        assert result["messages"][0].tool_calls[0]["name"] == "search_codebase"
 
@pytest.mark.asyncio
async def test_infinite_loop_prevention():
    """Test that infinite loop prevention works correctly"""
    state = AgentState(
        messages=[HumanMessage(content="test")],
        current_step="agent",
        tool_results={},
        iteration_count=10,  # Already at max
        error=None
    )
 
    mock_response = AIMessage(
        content="",
        tool_calls=[{"name": "search_codebase", "args": {}, "id": "call_999"}]
    )
 
    with patch.object(llm_with_tools, "invoke", return_value=mock_response):
        next_node = should_continue(state)
        # Should return "end" since iteration_count has reached the limit
        assert next_node == "end"
 
@pytest.mark.asyncio
async def test_checkpoint_persistence():
    """Test checkpoint persistence using in-memory SQLite"""
    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
 
    async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
        test_app = graph.compile(checkpointer=checkpointer)
        config = {"configurable": {"thread_id": "test-thread-001"}}
 
        # First run
        await test_app.ainvoke(
            {"messages": [HumanMessage(content="Hello")]},
            config=config
        )
 
        # Verify checkpoint was saved
        state = await test_app.aget_state(config)
        assert state is not None
        assert len(state.values["messages"]) > 0

from langchain_core.messages import trim_messages, SystemMessage
 
def trim_agent_messages(state: AgentState) -> dict:
    """Trim message history to reduce token usage"""
 
    trimmed = trim_messages(
        state["messages"],
        max_tokens=4096,          # A good target is ~50% of the model's input limit
        strategy="last",          # Keep the most recent messages
        token_counter=llm,        # Use the model's own tokenizer for counting
        include_system=True,      # Always preserve system messages
        allow_partial=False,      # Never split a message mid-way
        start_on="human"          # Ensure the history starts with a HumanMessage
    )
 
    return {"messages": trimmed}
 
# Add the trimming node to the graph
graph.add_node("trim_history", trim_agent_messages)
graph.add_edge("tools", "trim_history")    # Trim after each tool execution
graph.add_edge("trim_history", "agent")    # Then return to agent

import hashlib
import json
from datetime import datetime, timedelta
 
class CachedToolWrapper:
    """Tool wrapper with TTL-based caching"""
 
    def __init__(self, tool_func, ttl_minutes: int = 10):
        self.tool_func = tool_func
        self.ttl = timedelta(minutes=ttl_minutes)
        self._cache: dict = {}
 
    def _make_cache_key(self, **kwargs) -> str:
        """Generate a deterministic cache key from arguments"""
        return hashlib.md5(json.dumps(kwargs, sort_keys=True).encode()).hexdigest()
 
    def __call__(self, **kwargs) -> str:
        cache_key = self._make_cache_key(**kwargs)
        now = datetime.now()
 
        # Check cache
        if cache_key in self._cache:
            result, timestamp = self._cache[cache_key]
            if now - timestamp < self.ttl:
                print(f"Cache hit for {self.tool_func.__name__}")
                return result
 
        # Cache miss: execute the tool
        result = self.tool_func(**kwargs)
        self._cache[cache_key] = (result, now)
        return result
 
# Create cached version of expensive tools
cached_search = CachedToolWrapper(search_codebase, ttl_minutes=5)

from langchain_google_genai import ChatGoogleGenerativeAI
 
class AdaptiveModelSelector:
    """Dynamically select the optimal model based on task complexity"""
 
    def __init__(self):
        # Cost-efficient (for simpler tasks)
        self.flash_model = ChatGoogleGenerativeAI(
            model="gemini-2.0-flash",
            temperature=0
        )
        # High-precision (for complex tasks)
        self.pro_model = ChatGoogleGenerativeAI(
            model="gemini-2.0-pro",
            temperature=0
        )
 
    def select_model(self, task_complexity: str, requires_reasoning: bool = False):
        """Return the best model for the given task characteristics"""
        if requires_reasoning or task_complexity == "high":
            return self.pro_model
        return self.flash_model
 
# Apply adaptive model selection in the agent node
def adaptive_agent_node(state: AgentState) -> dict:
    """Agent node that switches models based on complexity signals"""
    selector = AdaptiveModelSelector()
 
    iteration = state.get("iteration_count", 0)
    is_complex = iteration > 3 or "architecture" in str(state["messages"]).lower()
 
    model = selector.select_model(
        task_complexity="high" if is_complex else "low",
        requires_reasoning=is_complex
    )
 
    model_with_tools = model.bind_tools(tools)
    response = model_with_tools.invoke(state["messages"])
 
    return {
        "messages": [response],
        "current_step": "agent",
        "iteration_count": iteration + 1
    }

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
import uuid
from typing import Optional
 
api_server = FastAPI(title="LangGraph Agent API")
 
# In-memory task registry (use Redis or Postgres in production)
task_registry: dict = {}
 
class TaskCreateRequest(BaseModel):
    task_description: str
    user_id: str
    priority: str = "normal"
 
class TaskStatusResponse(BaseModel):
    task_id: str
    status: str           # running / waiting_approval / completed / failed
    current_step: str
    result: Optional[str] = None
    waiting_for: Optional[str] = None
 
@api_server.post("/tasks", response_model=dict)
async def create_task(
    request: TaskCreateRequest,
    background_tasks: BackgroundTasks
):
    """Start a new agent task"""
    task_id = str(uuid.uuid4())
 
    task_registry[task_id] = {
        "status": "running",
        "current_step": "start",
        "result": None,
        "error": None
    }
 
    background_tasks.add_task(
        run_agent_task,
        task_id=task_id,
        description=request.task_description,
        thread_id=f"{request.user_id}-{task_id}"
    )
 
    return {"task_id": task_id, "status": "started"}
 
async def run_agent_task(task_id: str, description: str, thread_id: str):
    """Execute the agent task in the background"""
    config = {"configurable": {"thread_id": thread_id}}
 
    try:
        result = await app.ainvoke(
            {
                "messages": [HumanMessage(content=description)],
                "current_step": "start",
                "tool_results": {},
                "iteration_count": 0,
                "error": None
            },
            config=config
        )
 
        # Check if agent paused at an interrupt_before node
        state = await app.aget_state(config)
        if state.next:  # Remaining nodes = waiting for approval
            task_registry[task_id].update({
                "status": "waiting_approval",
                "current_step": state.next[0],
                "waiting_for": f"Approval required for: {state.next[0]}"
            })
        else:
            task_registry[task_id].update({
                "status": "completed",
                "result": result["messages"][-1].content
            })
 
    except Exception as e:
        task_registry[task_id].update({
            "status": "failed",
            "error": str(e)
        })
 
@api_server.get("/tasks/{task_id}/status", response_model=TaskStatusResponse)
async def get_task_status(task_id: str):
    """Get the current status of a task"""
    if task_id not in task_registry:
        raise HTTPException(status_code=404, detail="Task not found")
 
    task = task_registry[task_id]
    return TaskStatusResponse(
        task_id=task_id,
        status=task["status"],
        current_step=task.get("current_step", "unknown"),
        result=task.get("result"),
        waiting_for=task.get("waiting_for")
    )
 
@api_server.post("/tasks/{task_id}/approve")
async def approve_task(task_id: str, background_tasks: BackgroundTasks):
    """Resume a task that is waiting for approval"""
    if task_id not in task_registry:
        raise HTTPException(status_code=404, detail="Task not found")
 
    if task_registry[task_id]["status"] != "waiting_approval":
        raise HTTPException(status_code=400, detail="Task is not waiting for approval")
 
    task_registry[task_id]["status"] = "running"
    background_tasks.add_task(resume_agent_task, task_id=task_id)
    return {"message": "Task resumed", "task_id": task_id}

from contextlib import asynccontextmanager
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage application lifecycle for safe startup and shutdown"""
    print("🚀 Starting LangGraph Agent API Server...")
    await checkpointer.setup()
    print("✅ Checkpoint store initialized")
 
    yield  # Application runs here
 
    # On shutdown: safely handle in-flight tasks
    print("🔄 Shutting down gracefully...")
    running_tasks = [
        tid for tid, t in task_registry.items()
        if t["status"] == "running"
    ]
    if running_tasks:
        print(f"⚠️ {len(running_tasks)} tasks still running — state is saved in checkpoint store")
        # Tasks can be resumed after restart since state is persisted
 
    print("✅ Shutdown complete")
 
api_server = FastAPI(title="LangGraph Agent API", lifespan=lifespan)
 
@api_server.get("/health")
async def health_check():
    """Health check endpoint for load balancer probes"""
    return {
        "status": "healthy",
        "active_tasks": len([t for t in task_registry.values() if t["status"] == "running"]),
        "pending_approvals": len([t for t in task_registry.values() if t["status"] == "waiting_approval"])
    }

import asyncio
from typing import AsyncIterator
 
async def event_driven_agent_loop(queue_client, max_concurrent: int = 5):
    """Continuously process tasks from an event queue"""
    semaphore = asyncio.Semaphore(max_concurrent)
 
    async def process_event(event: dict):
        async with semaphore:
            task_id = event.get("task_id", str(uuid.uuid4()))
            thread_id = f"event-{task_id}"
            config = {"configurable": {"thread_id": thread_id}}
 
            print(f"Processing event: {task_id}")
 
            try:
                result = await app.ainvoke(
                    {
                        "messages": [HumanMessage(content=event["payload"])],
                        "current_step": "start",
                        "tool_results": {},
                        "iteration_count": 0,
                        "error": None
                    },
                    config=config
                )
 
                # Publish result to output queue
                await queue_client.publish(
                    topic="agent-results",
                    message={
                        "task_id": task_id,
                        "status": "completed",
                        "result": result["messages"][-1].content
                    }
                )
 
            except Exception as e:
                await queue_client.publish(
                    topic="agent-errors",
                    message={"task_id": task_id, "error": str(e)}
                )
 
    # Main event loop
    async for event in queue_client.consume(topic="agent-tasks"):
        asyncio.create_task(process_event(event))

# Example: Code analysis agent powered by Antigravity MCP tools
 
@tool
def read_file_content(file_path: str) -> str:
    """Read the contents of a file in the current project"""
    # This would connect to Antigravity's file reading capability via MCP
    with open(file_path, 'r') as f:
        return f.read()
 
@tool
def search_in_files(pattern: str, directory: str = ".") -> list:
    """Search for a pattern across project files"""
    # Connects to Antigravity's code search via MCP
    import subprocess
    result = subprocess.run(
        ["grep", "-rn", pattern, directory, "--include=*.py"],
        capture_output=True, text=True
    )
    return result.stdout.split('\n')[:20]  # Return first 20 matches
 
@tool
def run_command(command: str) -> dict:
    """Execute a shell command and return the output"""
    import subprocess
    result = subprocess.run(
        command.split(), capture_output=True, text=True, timeout=30
    )
    return {
        "stdout": result.stdout,
        "stderr": result.stderr,
        "returncode": result.returncode
    }
 
# Build a code analysis agent with these tools
code_analysis_tools = [read_file_content, search_in_files, run_command]
analysis_llm = llm.bind_tools(code_analysis_tools)