Lesson 16.10: Function Calling & Tool Use Mechanics

def calculate_total(price: float, quantity: int) -> float:
    return price * quantity

def get_order_status(order_id: str) -> dict:
    orders = {
        "ORD-1001": {
            "status": "shipped",
            "carrier": "Pathao",
            "estimated_delivery": "2026-10-16",
        },
        "ORD-1002": {
            "status": "processing",
            "carrier": None,
            "estimated_delivery": None,
        },
    }

    order = orders.get(order_id)
    if order is None:
        return {
            "found": False,
            "message": f"Order {order_id} was not found in our database.",
        }

    return {
        "found": True,
        "order_id": order_id,
        **order,
    }


# Test verification
print(get_order_status("ORD-1001"))
# Output: {'found': True, 'order_id': 'ORD-1001', 'status': 'shipped', ...}

tools = [
    {
        "type": "function",
        "name": "get_order_status",
        "description": (
            "Look up the current status, logistics carrier, and estimated "
            "delivery date of an order using its unique order ID."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "The unique order identifier, such as ORD-1001.",
                }
            },
            "required": ["order_id"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]

import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

client = OpenAI(api_key=api_key)

tools = [
    {
        "type": "function",
        "name": "get_order_status",
        "description": "Look up order status by order ID.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "Order ID, e.g. ORD-1001."}
            },
            "required": ["order_id"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]

response = client.responses.create(
    model=model,
    tools=tools,
    input="What is the current status of order ORD-1001?",
)

print(response.output)

import json

# Arguments string extracted from the model's function_call item
arguments_json = '{"order_id": "ORD-1001"}'

parsed_args = json.loads(arguments_json)
result = get_order_status(order_id=parsed_args["order_id"])

print("Function Output:", result)

from pydantic import BaseModel, ConfigDict, Field, ValidationError


class GetOrderStatusInput(BaseModel):
    model_config = ConfigDict(extra="forbid")

    order_id: str = Field(
        min_length=3,
        max_length=50,
        pattern=r"^ORD-[A-Z0-9-]+$",
        description="Must start with ORD- followed by uppercase alphanumerics.",
    )


# Valid arguments
valid_data = {"order_id": "ORD-1001"}
validated = GetOrderStatusInput.model_validate(valid_data)
print("Validated ID:", validated.order_id)  # 'ORD-1001'

# Invalid or malicious input
try:
    invalid_data = {"order_id": "DROP TABLE orders;--"}
    GetOrderStatusInput.model_validate(invalid_data)
except ValidationError as e:
    print("Rejected by Pydantic:", e)

import json
import os
from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel, ConfigDict, Field, ValidationError

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

ORDERS = {
    "ORD-1001": {"status": "shipped", "carrier": "Pathao", "estimated_delivery": "2026-10-16"},
    "ORD-1002": {"status": "processing", "carrier": None, "estimated_delivery": None},
}


class GetOrderStatusInput(BaseModel):
    model_config = ConfigDict(extra="forbid")
    order_id: str = Field(min_length=3, max_length=50, pattern=r"^ORD-[A-Z0-9-]+$")


def get_order_status(order_id: str) -> dict:
    order = ORDERS.get(order_id)
    if not order:
        return {"found": False, "message": f"Order {order_id} not found."}
    return {"found": True, "order_id": order_id, **order}


TOOLS = [
    {
        "type": "function",
        "name": "get_order_status",
        "description": "Look up status and carrier of an order by order ID.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "Order ID like ORD-1001."}
            },
            "required": ["order_id"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]


def execute_tool(name: str, arguments_json: str) -> dict:
    if name != "get_order_status":
        return {"error": "unsupported_tool", "message": f"Tool {name} is not allowed."}

    try:
        raw_args = json.loads(arguments_json)
        validated = GetOrderStatusInput.model_validate(raw_args)
        return get_order_status(validated.order_id)
    except (json.JSONDecodeError, ValidationError) as err:
        return {"error": "invalid_arguments", "details": str(err)}
    except Exception:
        return {"error": "execution_failed", "details": "Internal service error."}


def ask_assistant(user_query: str) -> str:
    input_items = [{"role": "user", "content": user_query}]
    max_rounds = 5

    for _ in range(max_rounds):
        response = client.responses.create(
            model=MODEL,
            tools=TOOLS,
            input=input_items,
            instructions=(
                "You are an order support assistant. Use the order lookup tool when needed. "
                "Answer based on the tool result. Never invent order details."
            ),
        )

        # Preserve the model's output in conversation history
        input_items += response.output

        tool_calls = [item for item in response.output if item.type == "function_call"]

        # Terminate loop when no further tool calls are requested
        if not tool_calls:
            return response.output_text

        # Execute each proposed tool call and inject results
        for call in tool_calls:
            result = execute_tool(name=call.name, arguments_json=call.arguments)
            input_items.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result, ensure_ascii=False),
            })

    raise RuntimeError("Exceeded maximum tool rounds without resolution.")


if __name__ == "__main__":
    answer = ask_assistant("What is the current status of my order ORD-1001?")
    print("AI Answer:", answer)

[
  {"name": "get_order_status", "arguments": "{\"order_id\":\"ORD-1001\"}", "call_id": "call_1"},
  {"name": "get_order_status", "arguments": "{\"order_id\":\"ORD-1002\"}", "call_id": "call_2"}
]

from pydantic import BaseModel, ConfigDict, Field


class OrderLookupInput(BaseModel):
    model_config = ConfigDict(extra="forbid")

    order_id: str = Field(
        min_length=3,
        max_length=50,
        pattern=r"^ORD-[A-Z0-9-]+$",
        description="Standard Byteion order ID (e.g. ORD-1001).",
    )

from typing import Any, Dict

# Demo data store (replace with production database repository)
DATABASE: Dict[str, Dict[str, Any]] = {
    "ORD-1001": {
        "status": "shipped",
        "carrier": "Pathao",
        "tracking_code": "PTH-88419",
        "estimated_delivery": "2026-10-16",
    },
    "ORD-1002": {
        "status": "processing",
        "carrier": None,
        "tracking_code": None,
        "estimated_delivery": None,
    },
}


def query_order(order_id: str) -> Dict[str, Any]:
    record = DATABASE.get(order_id)
    if record is None:
        return {"found": False, "message": f"Order {order_id} does not exist."}
    return {"found": True, "order_id": order_id, **record}

import json
from typing import Any, Dict
from pydantic import ValidationError
from schemas import OrderLookupInput
from services.orders import query_order

TOOL_DEFINITIONS = [
    {
        "type": "function",
        "name": "get_order_status",
        "description": "Look up order status, shipping carrier, and estimated delivery by order ID.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "Order ID (e.g. ORD-1001)."}
            },
            "required": ["order_id"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]


def dispatch_tool(name: str, arguments_json: str) -> Dict[str, Any]:
    if name != "get_order_status":
        return {"error": "unknown_tool", "message": f"Tool '{name}' is not recognized."}

    try:
        raw_args = json.loads(arguments_json)
        validated = OrderLookupInput.model_validate(raw_args)
        return query_order(validated.order_id)
    except (json.JSONDecodeError, ValidationError) as err:
        return {"error": "invalid_arguments", "message": str(err)}
    except Exception:
        return {"error": "internal_error", "message": "Failed to query order service."}

import json
import os
from dotenv import load_dotenv
from openai import OpenAI
from tools import TOOL_DEFINITIONS, dispatch_tool

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")


def run_assistant(user_prompt: str) -> str:
    conversation = [{"role": "user", "content": user_prompt}]
    max_turns = 5

    for _ in range(max_turns):
        response = client.responses.create(
            model=MODEL,
            tools=TOOL_DEFINITIONS,
            input=conversation,
            instructions=(
                "You are Byteion's AI Order Support Assistant. "
                "Use the get_order_status tool when an order ID is mentioned. "
                "Synthesize the tool output clearly. "
                "If not found, clearly state that the order was not found."
            ),
        )

        conversation += response.output
        tool_calls = [item for item in response.output if item.type == "function_call"]

        if not tool_calls:
            return response.output_text

        for tool_call in tool_calls:
            result = dispatch_tool(tool_call.name, tool_call.arguments)
            conversation.append({
                "type": "function_call_output",
                "call_id": tool_call.call_id,
                "output": json.dumps(result, ensure_ascii=False),
            })

    raise RuntimeError("Exceeded maximum tool rounds.")


if __name__ == "__main__":
    query = "What is the status of my order ORD-1001?"
    print("User:", query)
    print("Assistant:", run_assistant(query))

from tools import dispatch_tool


def test_valid_order_dispatch():
    res = dispatch_tool("get_order_status", '{"order_id": "ORD-1001"}')
    assert res["found"] is True
    assert res["status"] == "shipped"
    assert res["carrier"] == "Pathao"


def test_order_not_found():
    res = dispatch_tool("get_order_status", '{"order_id": "ORD-9999"}')
    assert res["found"] is False


def test_invalid_arguments_rejected():
    res = dispatch_tool("get_order_status", '{"order_id": "BAD"}')
    assert "error" in res
    assert res["error"] == "invalid_arguments"


def test_unknown_tool_rejected():
    res = dispatch_tool("delete_database", "{}")
    assert res["error"] == "unknown_tool"