Sync from GitHub via hub-sync

.env.example

@@ -0,0 +1,2 @@
+HF_TOKEN=your_huggingface_token_here
+MODEL_ID=openai/gpt-oss-120b:fastest

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Christopher Mendoza
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,15 +1,309 @@
----
-title: 07 Tool Using Agent
-emoji: 🐢
-colorFrom: blue
-colorTo: red
-sdk: gradio
-sdk_version: 6.24.0
-python_version: '3.13'
-app_file: app.py
-pinned: false
-license: mit
-short_description: 07-tool-using-agent
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Tool-Using Operations Agent — Structured Tool Router
+
+This project demonstrates a controlled tool-using AI agent in which the language model may propose tool calls, but application code retains authority over:
+
+- which tools exist
+- which arguments are permitted
+- argument validation
+- tool execution
+- failure handling
+- result normalization
+- audit logging
+
+The central architectural principle is:
+
+The model proposes. The application validates and executes.
+
+## Project Summary
+
+This repository presents a practical example of a structured tool router for an operations-oriented agent. Rather than giving the model unrestricted access to Python, shell, SQL, filesystem, or operating-system execution, the system exposes only a small set of approved tools with explicit schemas and controlled validation.
+
+The agent is built with Python, Pydantic, an OpenAI-compatible client for Hugging Face Inference Providers, Gradio, SQLite, and the World Bank Countries API. The application records tool execution metadata in a JSONL audit log while keeping the public interface limited to the current request's tool activity.
+
+## Architecture
+
+```text
+User Request
+     ↓
+LLM / Tool Selection
+     ↓
+Structured ToolCall
+     ↓
+Tool Registry / Allowlist
+     ↓
+Argument Validation
+     ↓
+Controlled Executor
+     ↓
+Approved Tool
+     ↓
+Normalized ToolResult
+     ↓
+Audit Log
+     ↓
+Result Returned to LLM
+     ↓
+Final Answer
+```
+
+The LLM never receives arbitrary Python, shell, SQL, filesystem, or operating-system execution capability. All privileged actions are mediated by the application layer before a tool is executed.
+
+## Current Stack
+
+- Python
+- Pydantic
+- OpenAI-compatible Python client
+- Hugging Face Inference Providers
+- `openai/gpt-oss-120b:cerebras`
+- Gradio
+- SQLite
+- World Bank Countries API
+- pytest
+
+The Hugging Face credential is supplied through the environment variable:
+
+```bash
+HF_TOKEN
+```
+
+## Approved Tools
+
+### calculator
+
+A constrained deterministic arithmetic tool.
+
+Important security characteristics:
+
+- does not use unrestricted Python `eval()`
+- parses expressions through Python AST
+- permits only approved mathematical operations
+
+### search_inventory
+
+A constrained SQLite inventory lookup.
+
+Important security characteristics:
+
+- does not expose arbitrary SQL execution
+- accepts only approved search parameters
+- uses parameterized SQL
+- initializes demo inventory data when needed
+
+### lookup_country
+
+Retrieves structured country information using the World Bank Countries API.
+
+Important reliability characteristics:
+
+- controlled HTTP timeout
+- HTTP error handling
+- response-shape validation
+- normalized structured result
+- no API key required for the World Bank lookup itself
+
+## Agent Execution Flow
+
+The agent runs in a tool loop. The model may propose a tool call, but the application validates the call and executes it through a controlled executor. Tool results are normalized and returned to the model in a structured form, which allows the model to decide whether additional tool calls are needed before delivering a final answer.
+
+This design keeps the model largely responsible for reasoning and planning while the application remains responsible for trust boundaries and operational control.
+
+## Multi-Tool Behavior
+
+The project supports sequential tool orchestration. A request such as:
+
+```text
+How many Electronics items are in inventory, and what is 347 multiplied by that number?
+```
+
+can be handled in sequence:
+
+1. the model selects `search_inventory`
+2. it receives a count of 3 Electronics records
+3. the model selects `calculator`
+4. it calculates `347 * 3`
+5. it returns `1041`
+
+This is not a hardcoded workflow. The model chooses tools based on the request and the available tool schemas, while the application enforces safety and valid execution.
+
+## Auditability
+
+Every executed tool call is recorded with structured metadata:
+
+- a call ID is assigned
+- tool arguments are preserved
+- status is recorded as success or error
+- output or error is recorded
+- start and completion timestamps are recorded
+- execution duration is recorded
+
+Audit entries are persisted as JSONL in:
+
+```text
+logs/tool_audit.jsonl
+```
+
+The public Gradio interface is privacy-conscious in a portfolio/demo sense:
+
+- the complete audit log remains server-side
+- the public UI displays only audit records associated with the current request
+- historical calls from other visitors are not exposed through the current-run audit view
+
+This is not presented as a multi-tenant security platform; it is a privacy-conscious demo design intended for a single local or deployed app instance.
+
+## Failure Handling
+
+The executor normalizes tool failures into structured results instead of allowing unhandled exceptions to break the agent loop. Examples of handled failure modes include:
+
+- unknown or unregistered tool
+- missing required argument
+- unexpected argument
+- wrong argument type
+- division by zero
+- external API failure
+
+This keeps the model operating on consistent result objects, even when a tool fails or the dependency is unavailable.
+
+## Repository Structure
+
+```text
+07-tool-using-agent/
+├── app.py
+├── requirements.txt
+├── .env.example
+├── data/
+│   └── .gitkeep
+├── logs/
+│   └── .gitkeep
+├── src/
+│   ├── __init__.py
+│   ├── agent.py
+│   ├── schemas.py
+│   ├── tool_registry.py
+│   ├── executor.py
+│   ├── audit.py
+│   └── tools/
+│       ├── __init__.py
+│       ├── calculator.py
+│       ├── database.py
+│       └── external_api.py
+└── tests/
+    ├── __init__.py
+    ├── test_tools.py
+    └── test_executor.py
+```
+
+Runtime-generated artifacts such as `operations.db` and `tool_audit.jsonl` are not presented here as tracked repository files.
+
+## Setup
+
+```bash
+git clone https://github.com/wushuchris/07-tool-using-agent.git
+cd 07-tool-using-agent
+
+python -m venv .venv
+source .venv/bin/activate
+
+pip install -r requirements.txt
+```
+
+Create a local `.env` file with:
+
+```bash
+HF_TOKEN=your_huggingface_token_here
+```
+
+You may also optionally set:
+
+```bash
+MODEL_ID=openai/gpt-oss-120b:cerebras
+```
+
+Notes:
+
+- `.env` is gitignored
+- the real token must never be committed
+- the token should be created with only the Hugging Face Inference Provider permissions required to run the model
+
+## Running
+
+```bash
+python app.py
+```
+
+This starts the Gradio interface for the agent.
+
+## Example Requests
+
+- `What is 347 multiplied by 29?`
+- `What electronics are currently in inventory?`
+- `What is the capital, region, and income level of Japan?`
+- `How many Electronics items are in inventory, and what is 347 multiplied by that number?`
+
+## Testing
+
+```bash
+pytest -v
+```
+
+The repository currently reports:
+
+**15 automated tests currently pass.**
+
+The test suite covers:
+
+- calculator behavior and restricted execution
+- SQLite inventory search
+- mocked World Bank response handling
+- approved tool execution
+- unregistered tool rejection
+- missing arguments
+- unexpected arguments
+- incorrect argument types
+- normalized tool failures
+- execution timing metadata
+
+The tests do not imply the live LLM provider is called as part of the unit test suite.
+
+## Security and Guardrails
+
+This project is designed around a narrow, explicit trust boundary:
+
+- tool allowlisting
+- constrained tool interfaces
+- schema-based arguments
+- rejection of unknown tools
+- no unrestricted `eval`
+- no arbitrary SQL
+- no shell execution tool
+- environment-based secret management
+- runtime database and audit logs are gitignored
+- public audit display is scoped to the current run
+
+This should be understood as a disciplined portfolio/demo design, not as a production-certified security system.
+
+## Design Lessons
+
+This project demonstrates several engineering lessons:
+
+1. Tool selection and tool execution should be separate concerns.
+2. The model should not directly control privileged capabilities.
+3. Tool calls should use explicit schemas.
+4. Tool failures should become structured data.
+5. External dependencies require validation and timeout/error handling.
+6. Multi-step requests may require sequential tool calls.
+7. Tool activity should be observable and auditable.
+
+## Deployment
+
+The application is designed to be deployable to Hugging Face Spaces with:
+
+```bash
+HF_TOKEN
+```
+
+configured as a Space secret.
+
+
+## License
+
+This project is distributed under the repository's existing license terms.

app.py

@@ -0,0 +1,192 @@
+from typing import Any, Dict, List
+
+import gradio as gr
+
+from src.agent import run_agent
+from src.audit import read_audit_records
+from src.tool_registry import list_tools
+
+
+def get_run_audit_records(
+    call_ids: List[str],
+) -> List[Dict[str, Any]]:
+    """Return audit records belonging only to the current agent run."""
+
+    if not call_ids:
+        return []
+
+    records = read_audit_records()
+
+    matching_records = [
+        record
+        for record in records
+        if record.call.call_id in call_ids
+    ]
+
+    return [
+        record.model_dump(mode="json")
+        for record in matching_records
+    ]
+
+
+def handle_request(
+    user_request: str,
+):
+    """
+    Run the agent and return user-facing execution details.
+    """
+
+    if not user_request or not user_request.strip():
+        return (
+            "Please enter a request.",
+            [],
+            [],
+            [],
+        )
+
+    try:
+        result = run_agent(
+            user_request.strip()
+        )
+
+        tool_calls = [
+            call.model_dump(mode="json")
+            for call in result.tool_calls
+        ]
+
+        tool_results = [
+            tool_result.model_dump(mode="json")
+            for tool_result in result.tool_results
+        ]
+
+        call_ids = [
+            call.call_id
+            for call in result.tool_calls
+        ]
+
+        audit_records = get_run_audit_records(
+            call_ids
+        )
+
+        return (
+            result.final_answer,
+            tool_calls,
+            tool_results,
+            audit_records,
+        )
+
+    except Exception:
+        return (
+            (
+                "Agent execution failed. "
+                "Please review the server logs for details."
+            ),
+            [],
+            [],
+            [],
+        )
+
+
+APP_DESCRIPTION = """
+# Tool-Using Operations Agent
+
+A controlled AI agent that selects and executes approved tools through a
+validated execution layer.
+
+The language model **proposes** tool calls. Application code decides what
+tools exist, validates their arguments, executes them, captures failures,
+and records an audit trail.
+
+### Available tools
+
+- **calculator** — deterministic arithmetic
+- **search_inventory** — constrained SQLite inventory search
+- **lookup_country** — structured World Bank country lookup
+
+Try a single-tool request or a request that requires multiple tools.
+"""
+
+
+with gr.Blocks(
+    title="Tool-Using Operations Agent",
+) as demo:
+
+    gr.Markdown(APP_DESCRIPTION)
+
+    with gr.Row():
+        request_box = gr.Textbox(
+            label="Request",
+            placeholder=(
+                "Example: How many Electronics items are "
+                "in inventory, and what is 347 multiplied "
+                "by that number?"
+            ),
+            lines=3,
+        )
+
+    run_button = gr.Button(
+        "Run Agent",
+        variant="primary",
+    )
+
+    gr.Examples(
+        examples=[
+            [
+                "What is 347 multiplied by 29?"
+            ],
+            [
+                "What electronics are currently in inventory?"
+            ],
+            [
+                "What is the capital, region, and income level of Japan?"
+            ],
+            [
+                (
+                    "How many Electronics items are in inventory, "
+                    "and what is 347 multiplied by that number?"
+                )
+            ],
+        ],
+        inputs=request_box,
+    )
+
+    gr.Markdown("## Agent Response")
+
+    final_answer = gr.Textbox(
+        label="Final Answer",
+        lines=5,
+        interactive=False,
+    )
+
+    with gr.Row():
+        tool_calls_output = gr.JSON(
+            label="Tool Calls",
+        )
+
+        tool_results_output = gr.JSON(
+            label="Tool Results",
+        )
+
+    gr.Markdown(
+        "## Current Run Audit Trail\n"
+        "Only tool executions generated by this request are shown."
+    )
+
+    audit_output = gr.JSON(
+        label="Audit Records",
+    )
+
+    run_button.click(
+        fn=handle_request,
+        inputs=request_box,
+        outputs=[
+            final_answer,
+            tool_calls_output,
+            tool_results_output,
+            audit_output,
+        ],
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,6 @@
+gradio>=5.0,<6.0
+openai>=1.0,<2.0
+pydantic>=2.0,<3.0
+python-dotenv>=1.0,<2.0
+requests>=2.31,<3.0
+pytest>=8.0,<9.0

src/agent.py

@@ -0,0 +1,209 @@
+import json
+import os
+from typing import Any, Dict, List
+
+from dotenv import load_dotenv
+from openai import OpenAI
+
+from src.audit import write_audit_record
+from src.executor import execute_tool
+from src.schemas import AgentRunResult, ToolCall
+from src.tool_registry import get_tool_definitions
+
+
+load_dotenv(dotenv_path=".env")
+
+
+HF_BASE_URL = "https://router.huggingface.co/v1"
+
+DEFAULT_MODEL = "openai/gpt-oss-120b:cerebras"
+
+SYSTEM_PROMPT = """
+You are a controlled tool-using operations agent.
+
+You have access only to the tools explicitly provided to you.
+
+Rules:
+1. Use tools when they are needed to answer the user's request accurately.
+2. Do not claim that a tool was used unless you actually called it.
+3. Never invent tool results.
+4. Do not invent tools that are not available.
+5. Use the calculator for arithmetic rather than calculating mentally.
+6. Use search_inventory for questions about the inventory database.
+7. Use lookup_country for structured country information.
+8. You may call multiple tools when the user's request requires them.
+9. After receiving tool results, use those results to construct your answer.
+10. If a tool returns an error, do not hide the failure.
+11. Do not expose internal chain-of-thought. Provide concise conclusions only.
+"""
+
+
+def get_client() -> OpenAI:
+    """Create the Hugging Face OpenAI-compatible client."""
+
+    token = os.getenv("HF_TOKEN")
+
+    if not token:
+        raise RuntimeError(
+            "HF_TOKEN is not configured. "
+            "Set it as an environment variable before running the agent."
+        )
+
+    return OpenAI(
+        base_url=HF_BASE_URL,
+        api_key=token,
+    )
+
+
+def run_agent(
+    user_request: str,
+    model: str | None = None,
+    max_tool_rounds: int = 5,
+) -> AgentRunResult:
+    """
+    Run the tool-using agent.
+
+    The model may request approved tools. Every requested tool is passed
+    through the controlled executor and written to the audit log.
+    """
+
+    if not isinstance(user_request, str):
+        raise TypeError("user_request must be a string.")
+
+    user_request = user_request.strip()
+
+    if not user_request:
+        raise ValueError("user_request cannot be empty.")
+
+    if max_tool_rounds < 1:
+        raise ValueError(
+            "max_tool_rounds must be at least 1."
+        )
+
+    client = get_client()
+
+    selected_model = (
+        model
+        or os.getenv("MODEL_ID")
+        or DEFAULT_MODEL
+    )
+
+    tools = get_tool_definitions()
+
+    messages: List[Dict[str, Any]] = [
+        {
+            "role": "system",
+            "content": SYSTEM_PROMPT,
+        },
+        {
+            "role": "user",
+            "content": user_request,
+        },
+    ]
+
+    executed_calls: List[ToolCall] = []
+    executed_results = []
+
+    for _ in range(max_tool_rounds):
+        response = client.chat.completions.create(
+         model=selected_model,
+         messages=messages,
+         tools=tools,
+         tool_choice="auto",
+         reasoning_effort="low",
+        )
+
+        message = response.choices[0].message
+
+        assistant_message: Dict[str, Any] = {
+            "role": "assistant",
+            "content": message.content,
+        }
+
+        if message.tool_calls:
+            assistant_message["tool_calls"] = [
+                tool_call.model_dump()
+                for tool_call in message.tool_calls
+            ]
+
+        messages.append(assistant_message)
+
+        if not message.tool_calls:
+            final_answer = (
+                message.content
+                or "The agent returned no final answer."
+            )
+
+            return AgentRunResult(
+                user_request=user_request,
+                final_answer=final_answer,
+                tool_calls=executed_calls,
+                tool_results=executed_results,
+            )
+
+        for model_tool_call in message.tool_calls:
+            tool_name = model_tool_call.function.name
+
+            try:
+                arguments = json.loads(
+                    model_tool_call.function.arguments
+                )
+
+                if not isinstance(arguments, dict):
+                    raise ValueError(
+                        "Tool arguments must decode to an object."
+                    )
+
+            except (
+                json.JSONDecodeError,
+                ValueError,
+            ) as exc:
+                arguments = {
+                    "__invalid_arguments__": (
+                        model_tool_call.function.arguments
+                    )
+                }
+
+            call = ToolCall(
+                call_id=model_tool_call.id,
+                tool_name=tool_name,
+                arguments=arguments,
+            )
+
+            result = execute_tool(call)
+
+            write_audit_record(
+                call=call,
+                result=result,
+            )
+
+            executed_calls.append(call)
+            executed_results.append(result)
+
+            tool_payload = {
+                "status": result.status,
+                "output": result.output,
+                "error": result.error,
+            }
+
+            messages.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": model_tool_call.id,
+                    "content": json.dumps(
+                        tool_payload,
+                        default=str,
+                    ),
+                }
+            )
+
+    return AgentRunResult(
+        user_request=user_request,
+        final_answer=(
+            "The agent reached the maximum number "
+            "of tool-execution rounds before producing "
+            "a final answer."
+        ),
+        tool_calls=executed_calls,
+        tool_results=executed_results,
+    )

src/audit.py

@@ -0,0 +1,87 @@
+import json
+from pathlib import Path
+from typing import List
+
+from src.schemas import AuditRecord, ToolCall, ToolResult
+
+
+AUDIT_LOG_PATH = Path("logs/tool_audit.jsonl")
+
+
+def write_audit_record(
+    call: ToolCall,
+    result: ToolResult,
+) -> AuditRecord:
+    """
+    Write one tool execution record to the JSONL audit log.
+    """
+
+    AUDIT_LOG_PATH.parent.mkdir(
+        parents=True,
+        exist_ok=True,
+    )
+
+    record = AuditRecord(
+        call=call,
+        result=result,
+    )
+
+    with AUDIT_LOG_PATH.open(
+        "a",
+        encoding="utf-8",
+    ) as file:
+        file.write(
+            record.model_dump_json()
+            + "\n"
+        )
+
+    return record
+
+
+def read_audit_records() -> List[AuditRecord]:
+    """
+    Read all valid audit records from the JSONL log.
+    """
+
+    if not AUDIT_LOG_PATH.exists():
+        return []
+
+    records: List[AuditRecord] = []
+
+    with AUDIT_LOG_PATH.open(
+        "r",
+        encoding="utf-8",
+    ) as file:
+        for line in file:
+            line = line.strip()
+
+            if not line:
+                continue
+
+            try:
+                payload = json.loads(line)
+                records.append(
+                    AuditRecord.model_validate(
+                        payload
+                    )
+                )
+            except (
+                json.JSONDecodeError,
+                ValueError,
+            ):
+                # Ignore malformed lines rather than
+                # crashing the whole audit reader.
+                continue
+
+    return records
+
+
+def clear_audit_log() -> None:
+    """
+    Delete the audit log if it exists.
+
+    Primarily useful for tests and demo resets.
+    """
+
+    if AUDIT_LOG_PATH.exists():
+        AUDIT_LOG_PATH.unlink()

src/executor.py

@@ -0,0 +1,140 @@
+from datetime import datetime, timezone
+from time import perf_counter
+from typing import Any, Dict
+
+from src.schemas import ToolCall, ToolResult
+from src.tool_registry import TOOL_REGISTRY, get_tool
+
+
+def _validate_arguments(
+    tool_name: str,
+    arguments: Dict[str, Any],
+) -> None:
+    """
+    Validate tool arguments against the registry's parameter schema.
+
+    This is intentionally lightweight and supports the schema features
+    used by the tools in this project.
+    """
+
+    if tool_name not in TOOL_REGISTRY:
+        raise ValueError(
+            f"Tool '{tool_name}' is not registered."
+        )
+
+    schema = TOOL_REGISTRY[tool_name]["parameters"]
+
+    properties = schema.get("properties", {})
+    required = schema.get("required", [])
+    allow_additional = schema.get(
+        "additionalProperties",
+        True,
+    )
+
+    for required_argument in required:
+        if required_argument not in arguments:
+            raise ValueError(
+                f"Missing required argument "
+                f"'{required_argument}' for tool "
+                f"'{tool_name}'."
+            )
+
+    if not allow_additional:
+        unknown_arguments = (
+            set(arguments.keys()) - set(properties.keys())
+        )
+
+        if unknown_arguments:
+            unknown = ", ".join(
+                sorted(unknown_arguments)
+            )
+
+            raise ValueError(
+                f"Unexpected argument(s) for tool "
+                f"'{tool_name}': {unknown}"
+            )
+
+    python_types = {
+        "string": str,
+        "integer": int,
+        "number": (int, float),
+        "boolean": bool,
+        "object": dict,
+        "array": list,
+    }
+
+    for argument_name, argument_value in arguments.items():
+        parameter_schema = properties.get(argument_name)
+
+        if not parameter_schema:
+            continue
+
+        expected_type_name = parameter_schema.get("type")
+
+        if not expected_type_name:
+            continue
+
+        expected_python_type = python_types.get(
+            expected_type_name
+        )
+
+        if expected_python_type is None:
+            continue
+
+        if not isinstance(
+            argument_value,
+            expected_python_type,
+        ):
+            raise TypeError(
+                f"Argument '{argument_name}' for tool "
+                f"'{tool_name}' must be of type "
+                f"'{expected_type_name}'."
+            )
+
+
+def execute_tool(call: ToolCall) -> ToolResult:
+    """
+    Validate and execute a tool call.
+
+    All outcomes are normalized into ToolResult objects so callers do not
+    need separate handling for every underlying tool.
+    """
+
+    started_at = datetime.now(timezone.utc)
+    timer_start = perf_counter()
+
+    try:
+        _validate_arguments(
+            tool_name=call.tool_name,
+            arguments=call.arguments,
+        )
+
+        tool = get_tool(call.tool_name)
+
+        output = tool(**call.arguments)
+
+        status = "success"
+        error = None
+
+    except Exception as exc:
+        output = None
+        status = "error"
+        error = f"{type(exc).__name__}: {exc}"
+
+    timer_end = perf_counter()
+    completed_at = datetime.now(timezone.utc)
+
+    duration_ms = (
+        timer_end - timer_start
+    ) * 1000
+
+    return ToolResult(
+        call_id=call.call_id,
+        tool_name=call.tool_name,
+        status=status,
+        output=output,
+        error=error,
+        started_at=started_at,
+        completed_at=completed_at,
+        duration_ms=round(duration_ms, 3),
+    )

src/schemas.py

@@ -0,0 +1,45 @@
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Literal, Optional
+from uuid import uuid4
+
+from pydantic import BaseModel, Field
+
+
+class ToolCall(BaseModel):
+    """A validated request to execute one approved tool."""
+
+    call_id: str = Field(default_factory=lambda: str(uuid4()))
+    tool_name: str
+    arguments: Dict[str, Any] = Field(default_factory=dict)
+
+
+class ToolResult(BaseModel):
+    """Normalized result returned by every tool."""
+
+    call_id: str
+    tool_name: str
+    status: Literal["success", "error"]
+    output: Optional[Any] = None
+    error: Optional[str] = None
+    started_at: datetime
+    completed_at: datetime
+    duration_ms: float
+
+
+class AuditRecord(BaseModel):
+    """Persistent record of one tool execution."""
+
+    timestamp: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc)
+    )
+    call: ToolCall
+    result: ToolResult
+
+
+class AgentRunResult(BaseModel):
+    """Top-level result returned after an agent run."""
+
+    user_request: str
+    final_answer: str
+    tool_calls: List[ToolCall] = Field(default_factory=list)
+    tool_results: List[ToolResult] = Field(default_factory=list)

src/tool_registry.py

@@ -0,0 +1,122 @@
+from typing import Any, Callable, Dict
+
+from src.tools.calculator import calculator
+from src.tools.database import search_inventory
+from src.tools.external_api import lookup_country
+
+
+ToolFunction = Callable[..., Dict[str, Any]]
+
+
+TOOL_REGISTRY: Dict[str, Dict[str, Any]] = {
+    "calculator": {
+        "function": calculator,
+        "description": (
+            "Safely evaluate a mathematical expression. "
+            "Use this tool for arithmetic and calculations."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "expression": {
+                    "type": "string",
+                    "description": (
+                        "A mathematical expression such as "
+                        "'125 * 4' or '(100 / 4) + 7'."
+                    ),
+                }
+            },
+            "required": ["expression"],
+            "additionalProperties": False,
+        },
+    },
+
+    "search_inventory": {
+        "function": search_inventory,
+        "description": (
+            "Search the local inventory database by item name, "
+            "category, or both."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "item_name": {
+                    "type": "string",
+                    "description": (
+                        "Optional item name or partial item name."
+                    ),
+                },
+                "category": {
+                    "type": "string",
+                    "description": (
+                        "Optional exact inventory category such as "
+                        "'Electronics', 'Furniture', or "
+                        "'Office Supplies'."
+                    ),
+                },
+            },
+            "additionalProperties": False,
+        },
+    },
+
+    "lookup_country": {
+        "function": lookup_country,
+        "description": (
+            "Look up structured information about a country "
+            "using the World Bank Countries API."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "country_name": {
+                    "type": "string",
+                    "description": (
+                        "The country name to look up, such as "
+                        "'Japan' or 'Canada'."
+                    ),
+                }
+            },
+            "required": ["country_name"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+
+def get_tool(name: str) -> ToolFunction:
+    """Return an approved tool function by name."""
+
+    if name not in TOOL_REGISTRY:
+        raise ValueError(
+            f"Tool '{name}' is not registered."
+        )
+
+    return TOOL_REGISTRY[name]["function"]
+
+
+def get_tool_definitions() -> list[Dict[str, Any]]:
+    """
+    Return tool definitions in OpenAI-compatible function-call format.
+    """
+
+    definitions = []
+
+    for name, config in TOOL_REGISTRY.items():
+        definitions.append(
+            {
+                "type": "function",
+                "function": {
+                    "name": name,
+                    "description": config["description"],
+                    "parameters": config["parameters"],
+                },
+            }
+        )
+
+    return definitions
+
+
+def list_tools() -> list[str]:
+    """Return the names of all approved tools."""
+
+    return sorted(TOOL_REGISTRY.keys())

src/tools/calculator.py

@@ -0,0 +1,88 @@
+import ast
+import operator
+from typing import Any, Dict
+
+
+_ALLOWED_OPERATORS = {
+    ast.Add: operator.add,
+    ast.Sub: operator.sub,
+    ast.Mult: operator.mul,
+    ast.Div: operator.truediv,
+    ast.FloorDiv: operator.floordiv,
+    ast.Mod: operator.mod,
+    ast.Pow: operator.pow,
+    ast.USub: operator.neg,
+    ast.UAdd: operator.pos,
+}
+
+
+def _evaluate(node: ast.AST) -> float:
+    if isinstance(node, ast.Expression):
+        return _evaluate(node.body)
+
+    if isinstance(node, ast.Constant):
+        if isinstance(node.value, (int, float)):
+            return node.value
+        raise ValueError("Only numeric constants are allowed.")
+
+    if isinstance(node, ast.BinOp):
+        operator_type = type(node.op)
+
+        if operator_type not in _ALLOWED_OPERATORS:
+            raise ValueError(
+                f"Operator {operator_type.__name__} is not allowed."
+            )
+
+        left = _evaluate(node.left)
+        right = _evaluate(node.right)
+
+        return _ALLOWED_OPERATORS[operator_type](left, right)
+
+    if isinstance(node, ast.UnaryOp):
+        operator_type = type(node.op)
+
+        if operator_type not in _ALLOWED_OPERATORS:
+            raise ValueError(
+                f"Operator {operator_type.__name__} is not allowed."
+            )
+
+        value = _evaluate(node.operand)
+
+        return _ALLOWED_OPERATORS[operator_type](value)
+
+    raise ValueError(
+        f"Unsupported expression element: {type(node).__name__}"
+    )
+
+
+def calculator(expression: str) -> Dict[str, Any]:
+    """
+    Safely evaluate a basic mathematical expression.
+
+    Supported operations:
+    +, -, *, /, //, %, **, unary +, unary -
+    """
+
+    if not isinstance(expression, str):
+        raise TypeError("expression must be a string.")
+
+    expression = expression.strip()
+
+    if not expression:
+        raise ValueError("expression cannot be empty.")
+
+    if len(expression) > 200:
+        raise ValueError("expression is too long.")
+
+    try:
+        parsed = ast.parse(expression, mode="eval")
+        result = _evaluate(parsed)
+    except ZeroDivisionError as exc:
+        raise ValueError("Division by zero is not allowed.") from exc
+    except SyntaxError as exc:
+        raise ValueError("Invalid mathematical expression.") from exc
+
+    return {
+        "expression": expression,
+        "result": result,
+    }

src/tools/database.py

@@ -0,0 +1,107 @@
+import sqlite3
+from pathlib import Path
+from typing import Any, Dict, List
+
+
+DB_PATH = Path("data/operations.db")
+
+
+def initialize_database() -> None:
+    """Create the demo database and seed it if needed."""
+
+    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+
+    with sqlite3.connect(DB_PATH) as conn:
+        cursor = conn.cursor()
+
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS inventory (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                item_name TEXT NOT NULL,
+                category TEXT NOT NULL,
+                quantity INTEGER NOT NULL,
+                unit_price REAL NOT NULL
+            )
+            """
+        )
+
+        cursor.execute("SELECT COUNT(*) FROM inventory")
+        row_count = cursor.fetchone()[0]
+
+        if row_count == 0:
+            sample_rows = [
+                ("Laptop", "Electronics", 12, 1299.99),
+                ("Monitor", "Electronics", 25, 349.50),
+                ("Keyboard", "Electronics", 40, 89.99),
+                ("Office Chair", "Furniture", 8, 499.00),
+                ("Desk", "Furniture", 6, 799.00),
+                ("Notebook", "Office Supplies", 120, 4.99),
+                ("Pen Set", "Office Supplies", 75, 12.50),
+            ]
+
+            cursor.executemany(
+                """
+                INSERT INTO inventory (
+                    item_name,
+                    category,
+                    quantity,
+                    unit_price
+                )
+                VALUES (?, ?, ?, ?)
+                """,
+                sample_rows,
+            )
+
+        conn.commit()
+
+
+def search_inventory(
+    item_name: str = "",
+    category: str = "",
+) -> Dict[str, Any]:
+    """
+    Search inventory using optional item name and category filters.
+
+    This tool exposes a constrained query interface rather than allowing
+    arbitrary SQL execution.
+    """
+
+    initialize_database()
+
+    query = """
+        SELECT
+            id,
+            item_name,
+            category,
+            quantity,
+            unit_price
+        FROM inventory
+        WHERE 1 = 1
+    """
+
+    parameters: List[Any] = []
+
+    if item_name:
+        query += " AND LOWER(item_name) LIKE LOWER(?)"
+        parameters.append(f"%{item_name.strip()}%")
+
+    if category:
+        query += " AND LOWER(category) = LOWER(?)"
+        parameters.append(category.strip())
+
+    query += " ORDER BY item_name"
+
+    with sqlite3.connect(DB_PATH) as conn:
+        conn.row_factory = sqlite3.Row
+        cursor = conn.execute(query, parameters)
+        rows = [dict(row) for row in cursor.fetchall()]
+
+    return {
+        "filters": {
+            "item_name": item_name,
+            "category": category,
+        },
+        "count": len(rows),
+        "items": rows,
+    }

src/tools/external_api.py

@@ -0,0 +1,117 @@
+from typing import Any, Dict, List
+
+import requests
+
+
+BASE_URL = "https://api.worldbank.org/v2/country"
+
+
+def lookup_country(country_name: str) -> Dict[str, Any]:
+    """
+    Retrieve basic country information from the World Bank Countries API.
+
+    This tool demonstrates controlled access to an external HTTP API
+    without requiring an API key.
+    """
+
+    if not isinstance(country_name, str):
+        raise TypeError("country_name must be a string.")
+
+    country_name = country_name.strip()
+
+    if not country_name:
+        raise ValueError("country_name cannot be empty.")
+
+    if len(country_name) > 100:
+        raise ValueError("country_name is too long.")
+
+    try:
+        response = requests.get(
+            BASE_URL,
+            params={
+                "format": "json",
+                "per_page": 400,
+            },
+            timeout=10,
+        )
+
+        response.raise_for_status()
+
+    except requests.Timeout as exc:
+        raise RuntimeError(
+            "World Bank API request timed out."
+        ) from exc
+
+    except requests.RequestException as exc:
+        raise RuntimeError(
+            f"World Bank API request failed: {exc}"
+        ) from exc
+
+    try:
+        payload = response.json()
+    except ValueError as exc:
+        raise RuntimeError(
+            "World Bank API returned invalid JSON."
+        ) from exc
+
+    if (
+        not isinstance(payload, list)
+        or len(payload) < 2
+        or not isinstance(payload[1], list)
+    ):
+        raise RuntimeError(
+            "World Bank API returned an unexpected response."
+        )
+
+    countries: List[Dict[str, Any]] = payload[1]
+
+    normalized_query = country_name.casefold()
+
+    exact_matches = [
+        country
+        for country in countries
+        if country.get("name", "").casefold() == normalized_query
+    ]
+
+    if exact_matches:
+        country = exact_matches[0]
+    else:
+        partial_matches = [
+            country
+            for country in countries
+            if normalized_query in country.get("name", "").casefold()
+        ]
+
+        if len(partial_matches) == 1:
+            country = partial_matches[0]
+        elif len(partial_matches) > 1:
+            names = [
+                match.get("name")
+                for match in partial_matches[:10]
+            ]
+
+            raise ValueError(
+                f"Country name '{country_name}' is ambiguous. "
+                f"Possible matches: {names}"
+            )
+        else:
+            raise ValueError(
+                f"No country found matching '{country_name}'."
+            )
+
+    region = country.get("region") or {}
+    income_level = country.get("incomeLevel") or {}
+    lending_type = country.get("lendingType") or {}
+
+    return {
+        "query": country_name,
+        "name": country.get("name"),
+        "iso2_code": country.get("iso2Code"),
+        "iso3_code": country.get("id"),
+        "capital": country.get("capitalCity"),
+        "region": region.get("value"),
+        "income_level": income_level.get("value"),
+        "lending_type": lending_type.get("value"),
+        "longitude": country.get("longitude"),
+        "latitude": country.get("latitude"),
+    }

tests/test_executor.py

@@ -0,0 +1,117 @@
+from src.executor import execute_tool
+from src.schemas import ToolCall
+
+
+def test_executor_runs_approved_tool():
+    call = ToolCall(
+        tool_name="calculator",
+        arguments={
+            "expression": "12 * 4",
+        },
+    )
+
+    result = execute_tool(call)
+
+    assert result.status == "success"
+    assert result.output is not None
+    assert result.output["result"] == 48
+    assert result.error is None
+    assert result.call_id == call.call_id
+    assert result.tool_name == "calculator"
+
+
+def test_executor_rejects_unregistered_tool():
+    call = ToolCall(
+        tool_name="run_shell_command",
+        arguments={
+            "command": "whoami",
+        },
+    )
+
+    result = execute_tool(call)
+
+    assert result.status == "error"
+    assert result.output is None
+    assert result.error is not None
+    assert "not registered" in result.error
+
+
+def test_executor_rejects_missing_required_argument():
+    call = ToolCall(
+        tool_name="calculator",
+        arguments={},
+    )
+
+    result = execute_tool(call)
+
+    assert result.status == "error"
+    assert result.output is None
+    assert result.error is not None
+    assert "Missing required argument" in result.error
+    assert "expression" in result.error
+
+
+def test_executor_rejects_unexpected_argument():
+    call = ToolCall(
+        tool_name="calculator",
+        arguments={
+            "expression": "10 + 5",
+            "unexpected": "value",
+        },
+    )
+
+    result = execute_tool(call)
+
+    assert result.status == "error"
+    assert result.output is None
+    assert result.error is not None
+    assert "Unexpected argument" in result.error
+    assert "unexpected" in result.error
+
+
+def test_executor_rejects_wrong_argument_type():
+    call = ToolCall(
+        tool_name="calculator",
+        arguments={
+            "expression": 123,
+        },
+    )
+
+    result = execute_tool(call)
+
+    assert result.status == "error"
+    assert result.output is None
+    assert result.error is not None
+    assert "must be of type 'string'" in result.error
+
+
+def test_executor_normalizes_tool_failure():
+    call = ToolCall(
+        tool_name="calculator",
+        arguments={
+            "expression": "10 / 0",
+        },
+    )
+
+    result = execute_tool(call)
+
+    assert result.status == "error"
+    assert result.output is None
+    assert result.error is not None
+    assert "Division by zero" in result.error
+
+
+def test_executor_records_timing_information():
+    call = ToolCall(
+        tool_name="calculator",
+        arguments={
+            "expression": "2 + 2",
+        },
+    )
+
+    result = execute_tool(call)
+
+    assert result.started_at is not None
+    assert result.completed_at is not None
+    assert result.completed_at >= result.started_at
+    assert result.duration_ms >= 0

tests/test_tools.py

@@ -0,0 +1,193 @@
+import pytest
+
+import src.tools.database as database_tool
+import src.tools.external_api as external_api_tool
+from src.tools.calculator import calculator
+
+
+def test_calculator_basic_math():
+    result = calculator("25 * 8")
+
+    assert result["expression"] == "25 * 8"
+    assert result["result"] == 200
+
+
+def test_calculator_respects_operator_precedence():
+    result = calculator("10 + 5 * 2")
+
+    assert result["result"] == 20
+
+
+def test_calculator_rejects_code_execution():
+    with pytest.raises(ValueError):
+        calculator(
+            "__import__('os').system('echo hacked')"
+        )
+
+
+def test_calculator_rejects_division_by_zero():
+    with pytest.raises(
+        ValueError,
+        match="Division by zero",
+    ):
+        calculator("10 / 0")
+
+
+def test_inventory_category_search(
+    tmp_path,
+    monkeypatch,
+):
+    test_db = tmp_path / "operations.db"
+
+    monkeypatch.setattr(
+        database_tool,
+        "DB_PATH",
+        test_db,
+    )
+
+    result = database_tool.search_inventory(
+        category="Electronics"
+    )
+
+    assert result["count"] == 3
+
+    names = {
+        item["item_name"]
+        for item in result["items"]
+    }
+
+    assert names == {
+        "Laptop",
+        "Monitor",
+        "Keyboard",
+    }
+
+
+def test_inventory_item_search(
+    tmp_path,
+    monkeypatch,
+):
+    test_db = tmp_path / "operations.db"
+
+    monkeypatch.setattr(
+        database_tool,
+        "DB_PATH",
+        test_db,
+    )
+
+    result = database_tool.search_inventory(
+        item_name="chair"
+    )
+
+    assert result["count"] == 1
+    assert (
+        result["items"][0]["item_name"]
+        == "Office Chair"
+    )
+
+
+class FakeResponse:
+    def __init__(
+        self,
+        payload,
+        status_code=200,
+    ):
+        self._payload = payload
+        self.status_code = status_code
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._payload
+
+
+def _world_bank_payload():
+    return [
+        {
+            "page": 1,
+            "pages": 1,
+            "total": 1,
+        },
+        [
+            {
+                "id": "JPN",
+                "iso2Code": "JP",
+                "name": "Japan",
+                "region": {
+                    "value": "East Asia & Pacific"
+                },
+                "incomeLevel": {
+                    "value": "High income"
+                },
+                "lendingType": {
+                    "value": "Not classified"
+                },
+                "capitalCity": "Tokyo",
+                "longitude": "139.77",
+                "latitude": "35.67",
+            }
+        ],
+    ]
+
+
+def test_country_lookup_success(
+    monkeypatch,
+):
+    def fake_get(*args, **kwargs):
+        return FakeResponse(
+            _world_bank_payload()
+        )
+
+    monkeypatch.setattr(
+        external_api_tool.requests,
+        "get",
+        fake_get,
+    )
+
+    result = external_api_tool.lookup_country(
+        "Japan"
+    )
+
+    assert result["name"] == "Japan"
+    assert result["iso2_code"] == "JP"
+    assert result["iso3_code"] == "JPN"
+    assert result["capital"] == "Tokyo"
+    assert (
+        result["region"]
+        == "East Asia & Pacific"
+    )
+    assert (
+        result["income_level"]
+        == "High income"
+    )
+
+
+def test_country_lookup_not_found(
+    monkeypatch,
+):
+    def fake_get(*args, **kwargs):
+        return FakeResponse(
+            [
+                {
+                    "page": 1,
+                    "pages": 1,
+                    "total": 0,
+                },
+                [],
+            ]
+        )
+
+    monkeypatch.setattr(
+        external_api_tool.requests,
+        "get",
+        fake_get,
+    )
+
+    with pytest.raises(
+        ValueError,
+        match="No country found",
+    ):
+        external_api_tool.lookup_country(
+            "DefinitelyNotARealCountryXYZ"
+        )