AI Agents & Tool Calling: Function Calling, Planning & Autonomous Workflows

{
  "name": "query_inventory",
  "description": "Look up available stock quantities, warehouse locations, and pricing for a specific product ID.",
  "parameters": {
    "type": "object",
    "properties": {
      "product_id": {
        "type": "integer",
        "description": "Unique numeric ID of the product (e.g. 1024)"
      },
      "include_discounts": {
        "type": "boolean",
        "description": "Whether to calculate active promotional wholesale discounts"
      }
    },
    "required": ["product_id"],
    "additionalProperties": false
  }
}

import ast
import operator

# ==========================================
# 1. TOOL DEFINITIONS & SAFE COMPUTATION
# ==========================================
# Safe arithmetic parser preventing dangerous eval() injection
SAFE_OPERATORS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.USub: operator.neg
}

def safe_eval_expr(node):
    if isinstance(node, ast.Constant):
        return node.value
    elif isinstance(node, ast.BinOp):
        left = safe_eval_expr(node.left)
        right = safe_eval_expr(node.right)
        op_type = type(node.op)
        if op_type in SAFE_OPERATORS:
            return SAFE_OPERATORS[op_type](left, right)
    elif isinstance(node, ast.UnaryOp):
        operand = safe_eval_expr(node.operand)
        op_type = type(node.op)
        if op_type in SAFE_OPERATORS:
            return SAFE_OPERATORS[op_type](operand)
    raise ValueError(f"Unsupported mathematical expression: {node}")

def calculate(expression: str) -> float:
    '''Safely evaluates a mathematical arithmetic expression string.'''
    tree = ast.parse(expression, mode='eval')
    return float(safe_eval_expr(tree.body))

# Mock Enterprise Inventory Database
INVENTORY_DB = {
    101: {"name": "Mechanical Keyboard", "stock": 25, "unit_price": 85.00},
    102: {"name": "Wireless Ergonomic Mouse", "stock": 42, "unit_price": 45.00},
    103: {"name": "27-inch 4K Monitor", "stock": 8, "unit_price": 320.00},
    104: {"name": "USB-C Multiport Hub", "stock": 60, "unit_price": 29.50}
}

def get_product_stock(product_id: int) -> dict:
    '''Retrieves stock level and unit price for a verified product ID.'''
    if not isinstance(product_id, int):
        raise TypeError("product_id must be an integer.")
    if product_id not in INVENTORY_DB:
        return {"error": f"Product ID {product_id} not found."}
    return INVENTORY_DB[product_id]

# ==========================================
# 2. TOOL REGISTRY & JSON SCHEMAS
# ==========================================
class ToolRegistry:
    def __init__(self):
        self._tools = {}
        self._schemas = []

    def register(self, name: str, func, schema: dict):
        self._tools[name] = func
        self._schemas.append(schema)

    def execute(self, name: str, arguments: dict):
        if name not in self._tools:
            raise KeyError(f"Tool '{name}' is not registered.")
        return self._tools[name](**arguments)

    def get_schemas(self):
        return self._schemas

# Initialize Registry
registry = ToolRegistry()

registry.register(
    name="calculate",
    func=calculate,
    schema={
        "name": "calculate",
        "description": "Perform mathematical arithmetic calculation.",
        "parameters": {
            "type": "object",
            "properties": {"expression": {"type": "string"}},
            "required": ["expression"]
        }
    }
)

registry.register(
    name="get_product_stock",
    func=get_product_stock,
    schema={
        "name": "get_product_stock",
        "description": "Look up stock inventory and unit price by product ID.",
        "parameters": {
            "type": "object",
            "properties": {"product_id": {"type": "integer"}},
            "required": ["product_id"]
        }
    }
)

# ==========================================
# 3. AGENT CONTROLLER LOOP
# ==========================================
class AgentController:
    def __init__(self, registry: ToolRegistry, max_steps: int = 5):
        self.registry = registry
        self.max_steps = max_steps

    def run_trace(self, simulated_tool_calls):
        '''Simulates the execution loop with observation feedback.'''
        messages = []
        print("=== Initiating Agent Execution Trace ===")

        for step in range(1, self.max_steps + 1):
            if not simulated_tool_calls:
                print(f"[Step {step}] No pending tool calls. Agent synthesizing final response.")
                return "Execution successfully concluded."

            call = simulated_tool_calls.pop(0)
            tool_name = call["name"]
            tool_args = call["arguments"]

            print(f"[Step {step}] Model requests Action: {tool_name}({tool_args})")

            # Execute tool safely
            try:
                observation = self.registry.execute(tool_name, tool_args)
                print(f"         Observation: {observation}")
                messages.append({"role": "tool", "name": tool_name, "content": str(observation)})
            except Exception as e:
                print(f"         Execution Error: {e}")
                messages.append({"role": "tool", "name": tool_name, "error": str(e)})

        print("[Warning] Agent terminated: max_steps threshold reached.")
        return "Aborted: step limit exceeded."

# Execute sample workflow
agent = AgentController(registry, max_steps=4)
simulated_calls = [
    {"name": "get_product_stock", "arguments": {"product_id": 101}},
    {"name": "calculate", "arguments": {"expression": "25 * 85.00"}}
]
agent.run_trace(simulated_calls)