104 lines
3.1 KiB
Python
104 lines
3.1 KiB
Python
"""Local tool-calling example with token-by-token streaming.
|
|
|
|
Flow:
|
|
1. Declare a `get_weather(city)` tool with a JSON Schema.
|
|
2. First turn: stream the chat. Print every text delta as it arrives;
|
|
feed each chunk into `ChatResult.add(chunk)` so tool_calls merge
|
|
correctly across the chunked wire format.
|
|
3. If the assistant called a tool, execute it locally and append a
|
|
`tool` message to history.
|
|
4. Second turn: stream again — this time the text reply is what we want
|
|
to see token-by-token.
|
|
|
|
Prerequisites: same as basic_usage.py (config.json, RAYCAST_BEARER,
|
|
RAYCAST_DEVICE_ID).
|
|
|
|
Run:
|
|
python examples/local_tool.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from raycast_api import ChatResult, Client, Config, Message, Tool
|
|
|
|
CONFIG_PATH = Path(os.environ.get("RAYCAST_CONFIG", "config.json"))
|
|
MODEL = os.environ.get("RAYCAST_MODEL", "Claude Sonnet 4.6")
|
|
|
|
WEATHER_TOOL = Tool.local(
|
|
name="get_weather",
|
|
description="Look up the current weather for a city.",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"city": {"type": "string", "description": "City name, e.g. 'Tokyo'."},
|
|
},
|
|
"required": ["city"],
|
|
},
|
|
)
|
|
|
|
|
|
def get_weather(city: str) -> dict[str, str]:
|
|
return {"city": city, "temp_c": "18", "conditions": "light rain"}
|
|
|
|
|
|
async def stream_turn(client: Client, history: list[Message]) -> ChatResult:
|
|
"""Stream one turn, printing text deltas live. Returns the merged result."""
|
|
result = ChatResult()
|
|
async for chunk in client.chat.stream(
|
|
model=MODEL, messages=history, tools=[WEATHER_TOOL]
|
|
):
|
|
if chunk.text:
|
|
print(chunk.text, end="", flush=True)
|
|
result.add(chunk)
|
|
print()
|
|
return result
|
|
|
|
|
|
async def main() -> None:
|
|
bearer = os.environ.get("RAYCAST_BEARER")
|
|
device_id = os.environ.get("RAYCAST_DEVICE_ID")
|
|
if not bearer or not device_id:
|
|
sys.exit("set RAYCAST_BEARER and RAYCAST_DEVICE_ID before running")
|
|
if not CONFIG_PATH.is_file():
|
|
sys.exit(f"no config at {CONFIG_PATH} — run `raycast-api init` first")
|
|
|
|
history: list[Message] = [
|
|
Message.user("What's the weather like in Tokyo right now?")
|
|
]
|
|
|
|
async with Client(
|
|
config=Config.load(CONFIG_PATH), bearer_token=bearer, device_id=device_id
|
|
) as client:
|
|
print("--- turn 1 ---")
|
|
first = await stream_turn(client, history)
|
|
|
|
if not first.tool_calls:
|
|
return
|
|
|
|
for tc in first.tool_calls:
|
|
print(f" → tool_call: {tc.name}({tc.arguments})")
|
|
|
|
history.append(first.to_assistant_message())
|
|
for call in first.tool_calls:
|
|
args = json.loads(call.arguments or "{}")
|
|
if call.name == "get_weather":
|
|
result: object = get_weather(**args)
|
|
else:
|
|
result = {"error": f"unknown tool: {call.name}"}
|
|
history.append(
|
|
Message.tool(tool_call_id=call.id, name=call.name, result=result)
|
|
)
|
|
|
|
print("--- turn 2 ---")
|
|
await stream_turn(client, history)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|