🤖 Single-Tool Agent#

Single-Tool Agent strips away all frameworks to reveal the exact mechanics of how an LLM calls a Python function using raw OpenAI Function Calling. On its own, an LLM has no idea what items your store sells or what they cost in rupees — that data lives in a private backend dictionary (PRICES).

When you ask "How much are the shoes?", the service executes a two-call Think → Act → Observe → Answer loop:

1. First LLM Call (Decision): Sends your question plus the JSON Schema menu card (TOOLS) describing get_price(item). Instead of guessing a price, the model replies with a structured tool_calls request: get_price(item="shoes").
2. Local Tool Execution: Python parses the JSON arguments, runs get_price("shoes") against PRICES to get "₹799", and appends a role="tool" message linked by tool_call_id.
3. Second LLM Call (Synthesis): Sends the full conversation (user question + assistant tool request + tool result) back to the model, which writes the final natural language reply.

1. Crucial Mental Model: The LLM Never Executes Code

A common misconception is that the cloud LLM reaches into your server to run get_price(). In reality, the LLM only generates text — when a tool is needed, it outputs a structured JSON object (msg.tool_calls) saying "I would like you to call function X with arguments Y". Your Python process is the gatekeeper that actually executes the function and decides what to return.

2. Anatomy of an OpenAI Tool Schema

The TOOLS list provides a JSON Schema contract for each function:

• name: The exact identifier ("get_price") the model will emit in call.function.name.
• description: The natural-language rule telling the model when this tool is appropriate ("Get the price of a shop item the user asks about.").
• parameters: A JSON Schema object specifying each argument's name, type ("string"), description, and required list.

3. Strict Message Ordering and tool_call_id

The Chat Completions API enforces a strict state-machine invariant on messages: every message with role="tool" must immediately follow an assistant message that contains a matching tool_calls[i].id. If you forget messages.append(msg) before appending the role="tool" result, the API rejects the second call with a 400 Bad Request error.

4. Conditional Routing (Tool vs. Direct Reply)

When you ask a general question like "Hello, who are you?", the model sees that get_price is irrelevant, leaves msg.tool_calls empty, and populates msg.content directly — completing in a single API call without touching the tool.

Core AI Module (src/python/03_agent_one_tool.py)

src/python/03_agent_one_tool.py implements a complete single-tool agent from scratch using raw OpenAI Function Calling. It defines a private catalog dictionary (PRICES), a local Python lookup function (get_price()), a JSON Schema tool specification (TOOLS), and the two-call agent decision loop (run_agent_with_tool()).

Schemas, Tool Definition, and Agent Decision Loop

• class AgentRequest(BaseModel) & class AgentResponse(BaseModel): Pydantic schemas defining the input query (message: str = "How much are the shoes?") and output payload (message: str, response: str).
• PRICES: dict[str, int]: A private product catalog ({"shoes": 799, "hat": 399, "bag": 1420, "shorts": 1299, "pants": 1699}) holding domain data that the LLM cannot know without invoking a tool.
• get_price(item: str) -> str: The local Python tool function. It normalizes item.lower(), looks up the price via PRICES.get(item.lower(), 'unknown'), and returns a rupee-formatted string such as "₹799" or "₹unknown".
• TOOLS: list[dict]: The JSON Schema specification passed to OpenAI's tools= parameter. It declares "type": "function", names the tool "get_price", explains when to call it via "description": "Get the price of a shop item the user asks about.", and requires a string parameter "item".
• run_agent_with_tool(question: str = "How much are the shoes?") -> str: Executes the Think → Act → Observe → Answer loop across seven numbered steps:
  1. Step ① — Initialize Client and Conversation History: Creates client = OpenAI(timeout=15.0), selects model = os.getenv("OPENAI_MODEL", "gpt-4o-mini"), and initializes messages = [{"role": "user", "content": question}].
  2. Step ② — Phase 1 (First LLM Call with Tool Schema): Calls msg = client.chat.completions.create(model=model, messages=messages, tools=TOOLS).choices[0].message so the model can decide whether to answer directly or request get_price.
  3. Step ③ — Phase 2 (Inspect msg.tool_calls): Checks if not msg.tool_calls:. When the question does not require a price lookup, the model leaves tool_calls empty and the function returns msg.content or "" immediately.
  4. Step ④ — Append Assistant Tool-Call Request: Appends msg to messages so the assistant's tool_calls entry precedes the corresponding tool outputs, satisfying the Chat Completions state-machine contract.
  5. Step ⑤ — Execute Local Function and Append role="tool" Output: Iterates over msg.tool_calls, parses call.function.arguments with json.loads(), runs get_price(args.get("item", "")) in Python, and appends {"role": "tool", "tool_call_id": call.id, "content": ...} to messages.
  6. Step ⑥ — Phase 3 (Second LLM Call for Final Synthesis): Calls client.chat.completions.create(model=model, messages=messages).choices[0].message with the full conversation history (user prompt + assistant tool call + tool output) so the LLM can craft a natural-language answer grounded in the exact price.
  7. Step ⑦ — Return Final Response: Returns msg.content or "".

How the AI Components Connect End-to-End

1. run_agent_with_tool(question) sends the user's question alongside the TOOLS JSON Schema to OpenAI (Call #1).
2. If the question asks about a shop item, the LLM responds with a structured tool_calls payload specifying get_price and JSON arguments {"item": "..."} instead of guessing a price.
3. Python appends the assistant's tool request to messages, executes get_price() against PRICES locally, and appends the result as a role="tool" message linked by tool_call_id.
4. The augmented messages transcript is sent back to OpenAI (Call #2), which reads the tool result and synthesizes the final natural-language answer.

Minimal Single-Tool Agent Demonstration.

"""Minimal Single-Tool Agent Demonstration.

This script demonstrates the foundational mechanics of an LLM Agent:
1. LLM + External Tool + Decision Loop (ReAct / Tool-Use Pattern).
2. The LLM does NOT execute Python code directly; instead, it returns structured
   JSON requesting a tool call when it needs external information.
3. The client application executes the requested function locally and feeds
   the result back to the LLM to generate the final response.
"""

import json
import os

from dotenv import find_dotenv, load_dotenv
from openai import OpenAI
from pydantic import BaseModel

# Load environment variables (e.g., OPENAI_API_KEY) from .env
load_dotenv(find_dotenv())


class AgentRequest(BaseModel):
    message: str = "How much are the shoes?"


class AgentResponse(BaseModel):
    message: str
    response: str


# ---------------------------------------------------------------------------
# Mock Database
# ---------------------------------------------------------------------------
# In a real-world system, this would be an external database, API, or service
# containing domain data that the LLM has no inherent knowledge of.
PRICES = {"shoes": 799, "hat": 399, "bag": 1420, "shorts": 1299, "pants": 1699}


# ---------------------------------------------------------------------------
# Step 1: Define the Python Tool Function
# ---------------------------------------------------------------------------
def get_price(item: str) -> str:
    """Fetch the price of a given shop item from the database.

    Args:
        item: Name of the item to look up.

    Returns:
        Formatted price string in INR, or '₹unknown' if item is not found.
    """
    # Log to console so tool invocations are visible during execution
    print(f"   🔧 tool called: get_price({item!r})")
    # Use .get() to safely handle unknown items without throwing a KeyError
    return f"₹{PRICES.get(item.lower(), 'unknown')}"


# ---------------------------------------------------------------------------
# Step 2: Define Tool Schema for the Model (JSON Schema / Function Calling)
# ---------------------------------------------------------------------------
# The model uses this schema to know:
# - Which tools are available ('name')
# - When to use each tool ('description' acts as guidance for the LLM)
# - What arguments to generate ('parameters' defines argument types and requirements)
TOOLS = [{
    "type": "function",
    "function": {
        "name": "get_price",
        "description": "Get the price of a shop item the user asks about.",
        "parameters": {
            "type": "object",
            "properties": {
                "item": {"type": "string", "description": "The item name, e.g. 'shoes' or 'bag'"},
            },
            "required": ["item"],
        },
    },
}]


# ---------------------------------------------------------------------------
# Step 3: The Agent Decision Loop (Think -> Tool Execution -> Final Response)
# ---------------------------------------------------------------------------
def run_agent_with_tool(question: str = "How much are the shoes?") -> str:
    """Process a user query, invoking tools dynamically if required.

    Execution Flow:
    1. Send user message and tool definitions to the LLM.
    2. Check if the LLM requested a tool call (msg.tool_calls).
    3. If requested:
       a. Append the assistant's tool-call request to message history.
       b. Parse arguments and execute the local Python function.
       c. Append the tool output with role='tool' and matching tool_call_id.
       d. Send full conversation history back to the LLM for final synthesis.
    4. If no tool requested, return the direct LLM text response.

    Args:
        question: Natural language prompt from the user.

    Returns:
        Final response text from the LLM.
    """
    # ① Initialize OpenAI client and conversation history with the user's prompt
    client = OpenAI(timeout=15.0)
    model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
    messages = [{"role": "user", "content": question}]

    # ② Phase 1: Send user message along with available tool definitions.
    # The model inspects the prompt and decides whether to respond directly
    # or request a tool invocation.
    msg = client.chat.completions.create(model=model, messages=messages, tools=TOOLS).choices[0].message

    # ③ Phase 2: Inspect model response to see if tool invocation was requested
    # (if not, return the model's direct answer immediately)
    if not msg.tool_calls:
        return msg.content or ""

    # ④ Step 2a: Crucial! Append the assistant's tool call message to history.
    # OpenAI requires the assistant's tool_calls message to precede tool responses.
    messages.append(msg)

    # ⑤ Step 2b: Iterate through requested tool calls, execute local function, and append result
    for call in msg.tool_calls:
        # Parse the JSON arguments generated by the model
        args = json.loads(call.function.arguments)

        # Invoke the corresponding Python function locally
        # Step 2c: Append tool result to history.
        # Role 'tool' and 'tool_call_id' link the output back to the model's call.
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": get_price(args.get("item", "")),
        })

    # ⑥ Phase 3: Second LLM call with complete history (Prompt + Tool Call + Tool Result).
    # The model reads the tool output and crafts a natural language response.
    msg = client.chat.completions.create(model=model, messages=messages).choices[0].message

    # ⑦ Return the assistant's final text content
    return msg.content or ""

Flask Blueprint server and route handler (agent_service) in src/python/app.py.

"""Flask server for 02-langchain-agents."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_summarize = _load_service("01_langchain_summarizer.py", "langchain_summarizer")
mod_memory = _load_service("02_langchain_memory_chat.py", "langchain_memory_chat")
mod_agent = _load_service("03_agent_one_tool.py", "agent_one_tool")
mod_lc_tool = _load_service("04_langchain_one_tool.py", "langchain_one_tool")
mod_lc_multi = _load_service("05_langchain_multi_tools.py", "langchain_multi_tools")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/summarize", methods=["POST"])
def summarizer_service():
    # ① Parse input text from JSON payload and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    text = (data.get("text") or data.get("message") or "").strip()
    if not text:
        return jsonify({"error": "Text cannot be empty"}), 400
    try:
        # ② Run the LangChain LCEL summarizer chain in 01_langchain_summarizer.py
        summary = mod_summarize.summarize(text)
        return jsonify({"result": summary, "summary": summary})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/memory-chat", methods=["POST"])
def memory_chat_service():
    # ① Parse user message and prior conversation history from JSON payload
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    history = data.get("history") or []
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Convert history dicts to ChatMessage models and run chat_with_memory()
        chat_history = [
            mod_memory.ChatMessage(**m) if isinstance(m, dict) else m
            for m in history
        ]
        reply, updated_history = mod_memory.chat_with_memory(message, chat_history)
        return jsonify({
            "result": reply,
            "response": reply,
            "history": [m.model_dump() if hasattr(m, "model_dump") else m for m in updated_history],
        })
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/agent", methods=["POST"])
def agent_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the raw OpenAI tool-calling loop in 03_agent_one_tool.py
        reply = mod_agent.run_agent_with_tool(message)
        return jsonify({"result": reply, "response": reply})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/langchain-one-tool", methods=["POST"])
def langchain_one_tool_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the LangChain single-tool agent in 04_langchain_one_tool.py
        reply, tool_called = mod_lc_tool.run_langchain_one_tool(message)
        return jsonify({"result": reply, "response": reply, "tool_called": tool_called})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/multi-tool-agent", methods=["POST"])
def multi_tool_agent_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the LangGraph ReAct multi-tool agent in 05_langchain_multi_tools.py
        reply, tool_called = mod_lc_multi.run_multi_tool_agent(message)
        return jsonify({"result": reply, "response": reply, "tool_called": tool_called})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Use this checklist to review the raw OpenAI tool-calling and agent loop mechanics demonstrated by the Single-Tool Agent service:

1. LLMs Request Tools; Python Executes Them: The LLM never runs Python code directly; when it needs external data, it emits a structured msg.tool_calls JSON payload asking your application to run a named function.
2. JSON Schema Tool Specification (TOOLS): Passing tools=TOOLS gives the model each function's name, natural-language description (when to call it), and JSON Schema parameters (what arguments to pass).
3. Two-Call ReAct Loop (Think → Act → Observe → Answer): Call #1 lets the model decide whether to invoke a tool; local Python runs the function; Call #2 feeds the tool output back so the model can write the final user-facing response.
4. Fast-Path Direct Replies: When msg.tool_calls is empty (e.g., for general knowledge questions), the loop short-circuits after Call #1 and immediately returns msg.content.
5. Strict Message Ordering Invariant: OpenAI requires you to append the assistant's tool-call message (messages.append(msg)) before appending the corresponding role="tool" result messages.
6. Correlation via tool_call_id: Every {"role": "tool", "tool_call_id": call.id, "content": ...} message must carry the exact call.id emitted by the model so the API can match each observation to its request.
7. Defensive Local Lookup: Normalizing arguments with item.lower() and using PRICES.get(..., "unknown") ensures unexpected model arguments return a graceful string instead of crashing the server with a KeyError.