GPT-6 Astra Function Calling 实战 2026：让 AI 调用 API、数据库和自定义工具

import json
from openai import OpenAI

client = OpenAI()

tools = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "Get current weather for a city via HTTP API.",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name, e.g. Tokyo, London, Shanghai",
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit. Default celsius.",
                },
            },
            "required": ["city"],
            "additionalProperties": False,
        },
    }
]

import json
import requests
from openai import OpenAI

client = OpenAI()

def call_weather_api(city: str, unit: str = "celsius") -> dict:
    """Replace with your real HTTP endpoint."""
    url = "https://api.example.com/weather"
    r = requests.get(url, params={"city": city, "unit": unit}, timeout=5)
    r.raise_for_status()
    return r.json()  # e.g. {"city": "Tokyo", "temp": 22, "condition": "sunny"}

# Initial request — model may call tools
response = client.responses.create(
    model="gpt-6-astra",
    input="What is the weather in Tokyo right now?",
    tools=tools,
    reasoning={"effort": "low"},
)

# Tool loop: keep going until model stops calling tools
while True:
    tool_calls = [o for o in response.output if o.type == "function_call"]
    if not tool_calls:
        break  # model is done — final answer is in response.output_text

    tool_results = []
    for call in tool_calls:
        args = json.loads(call.arguments)
        if call.name == "get_weather":
            result = call_weather_api(**args)
        else:
            result = {"error": f"Unknown tool: {call.name}"}
        tool_results.append({
            "type": "function_call_output",
            "call_id": call.call_id,
            "output": json.dumps(result, ensure_ascii=False),
        })

    # Append history + results and continue
    response = client.responses.create(
        model="gpt-6-astra",
        input=response.output + tool_results,
        tools=tools,
        reasoning={"effort": "low"},
    )

print(response.output_text)

import sqlite3
from typing import Any

# Only these parameterized queries can be executed — no raw user SQL
ALLOWED_QUERIES: dict[str, str] = {
    "get_order":   "SELECT id, status, total, created_at FROM orders WHERE id = ?",
    "list_orders": "SELECT id, status, total FROM orders WHERE user_id = ? LIMIT 20",
}

def query_db(query_name: str, params: list[Any]) -> list[dict]:
    """Execute an allowlisted, parameterized read-only query.
    Never accepts raw SQL strings from the model.
    """
    sql = ALLOWED_QUERIES.get(query_name)
    if sql is None:
        raise ValueError(f"Query '{query_name}' not in allowlist")
    con = sqlite3.connect("orders.db", check_same_thread=False)
    con.row_factory = sqlite3.Row
    try:
        rows = con.execute(sql, params).fetchall()
        return [dict(r) for r in rows]
    finally:
        con.close()

# Schema exposed to the model
db_tool = {
    "type": "function",
    "name": "query_db",
    "description": "Query the orders DB. Only read-only allowlisted queries.",
    "parameters": {
        "type": "object",
        "properties": {
            "query_name": {
                "type": "string",
                "enum": list(ALLOWED_QUERIES.keys()),
                "description": "Name of the allowlisted query",
            },
            "params": {
                "type": "array",
                "items": {"type": ["string", "number"]},
                "description": "Positional parameters for the query",
            },
        },
        "required": ["query_name", "params"],
    },
}