70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
"""Multi-turn: same backend, two turns, context persists across the pair.
|
|
|
|
The backend fingerprints `messages[:-1]` against its pool of live PTYs.
|
|
After turn 1 finishes, we re-send the conversation with the assistant's
|
|
reply appended and one new user message — fingerprint matches, the same
|
|
process is reused, and the server-side prompt cache keeps the context
|
|
without us paying a fresh-spawn tax.
|
|
|
|
Run:
|
|
python examples/multi_turn.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from claude_code_api import (
|
|
AssistantMessage,
|
|
BackendOptions,
|
|
ClaudeCodeBackend,
|
|
ResultMessage,
|
|
TextBlock,
|
|
)
|
|
|
|
CWD = Path(os.environ.get("CLAUDE_CODE_CWD", Path.cwd())).resolve()
|
|
|
|
|
|
def assistant_text(events: list[Any]) -> str:
|
|
for ev in reversed(events):
|
|
if isinstance(ev, AssistantMessage):
|
|
return "".join(b.text for b in ev.content if isinstance(b, TextBlock)).strip()
|
|
return ""
|
|
|
|
|
|
async def run_turn(
|
|
backend: ClaudeCodeBackend, history: list[dict[str, Any]]
|
|
) -> str:
|
|
events: list[Any] = []
|
|
async for event in backend.complete(history):
|
|
events.append(event)
|
|
if isinstance(event, ResultMessage):
|
|
print(f" → stop_reason={event.stop_reason} duration_ms={event.duration_ms}")
|
|
return assistant_text(events)
|
|
|
|
|
|
async def main() -> None:
|
|
opts = BackendOptions(cwd=str(CWD), dangerously_skip_permissions=True)
|
|
async with ClaudeCodeBackend(opts) as backend:
|
|
history: list[dict[str, Any]] = [
|
|
{"role": "user", "content": "Remember the codeword: Beaver. Reply OK."}
|
|
]
|
|
print("[turn 1]")
|
|
reply1 = await run_turn(backend, history)
|
|
print(f" [assistant] {reply1}")
|
|
|
|
history += [
|
|
{"role": "assistant", "content": [{"type": "text", "text": reply1}]},
|
|
{"role": "user", "content": "What was the codeword? Reply with one word."},
|
|
]
|
|
print("[turn 2]")
|
|
reply2 = await run_turn(backend, history)
|
|
print(f" [assistant] {reply2}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|