#!/usr/bin/env python3 """ LuckySt Trading Core — Shared trading intelligence for all agent interfaces. Extracted from telegram_bot.py so that Telegram, Swarm, and future adapters all share the same tool definitions, Claude system prompt, and execution logic. """ import os import json import uuid import logging import asyncio from pathlib import Path from datetime import datetime, timedelta, timezone from typing import Optional, Callable, Awaitable from dotenv import load_dotenv import httpx try: import anthropic HAS_ANTHROPIC = True except ImportError: HAS_ANTHROPIC = False # Load backend .env for defaults load_dotenv(Path(__file__).parent.parent / "backend" / ".env") logger = logging.getLogger("luckyst.core") # ============================================================================ # CONFIG # ============================================================================ API_BASE = os.environ.get("LUCKYST_API_BASE", "http://localhost:8000/api/v1") ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "") SESSION_DIR = Path(os.environ.get("LUCKBOT_SESSION_DIR", str(Path.home()))) SESSION_PATH = SESSION_DIR / ".openclawd_session.json" _session_lock = asyncio.Lock() # ============================================================================ # SESSION MANAGEMENT # ============================================================================ def load_session() -> dict: if SESSION_PATH.exists(): with open(SESSION_PATH) as f: return json.load(f) return {} def save_session(data: dict): SESSION_DIR.mkdir(parents=True, exist_ok=True) with open(SESSION_PATH, "w") as f: json.dump(data, f) os.chmod(SESSION_PATH, 0o600) # ============================================================================ # API CLIENT # ============================================================================ async def api_request(method: str, endpoint: str, data=None, token=None) -> dict: """Make authenticated request to the FastAPI backend.""" headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" url = f"{API_BASE}{endpoint}" async with httpx.AsyncClient(timeout=10) as client: if method == "GET": resp = await client.get(url, headers=headers) elif method == "POST": resp = await client.post(url, json=data, headers=headers) elif method == "DELETE": resp = await client.delete(url, headers=headers) else: return {"error": "Unknown method"} if resp.status_code >= 400: try: err = resp.json() return {"error": err.get("detail", f"HTTP {resp.status_code}")} except Exception: return {"error": f"HTTP {resp.status_code}"} if resp.status_code == 204: return {"ok": True} return resp.json() def get_token_and_creds(): """Get JWT token and Kalshi creds from session.""" session = load_session() return session.get("jwt_token"), session.get("kalshi_api_key"), session.get("rsa_key") def get_active_instance() -> int | None: return load_session().get("active_instance") # ============================================================================ # TRADING CONTEXT # ============================================================================ async def get_trading_context() -> str: """Fetch current trading state for AI context.""" session = load_session() token = session.get("jwt_token") if not token: return "Not logged in to terminal." try: result = await api_request("GET", "/terminal/auto/instances", token=token) if "error" in result: return "No active instances." instances = result.get("instances", []) if not instances: return "No active trading instances." lines = [] for inst in instances: markets = inst.get("markets", []) market_str = " vs ".join(m.split("-")[-1] for m in markets) lines.append( f"Instance #{inst['id']}: {market_str} | {inst.get('status', '?')} | " f"Pos: {inst.get('position', 0)} | P&L: {inst.get('pnl', '$0.00')}" ) return "Active instances:\n" + "\n".join(lines) except Exception: return "Could not fetch trading state." # ============================================================================ # CLAUDE TOOL-USE DEFINITIONS # ============================================================================ TRADING_TOOLS = [ { "name": "deploy_instance", "description": "Deploy a trading instance NOW on 1 or 2 Kalshi markets with JOIN market making. For 2 markets, bids NO on both sides. Use higher_first=true for basketball and other sports where filling the expensive side first gives positive time decay.", "input_schema": { "type": "object", "properties": { "markets": {"type": "array", "items": {"type": "string"}, "description": "1 or 2 Kalshi market tickers"}, "contract_size": {"type": "integer", "description": "Contracts per cycle (default 1)", "default": 1}, "higher_first": {"type": "boolean", "description": "Enable higher-first mode: trade the more expensive NO side first for positive time decay. Strongly recommended for basketball. Default false.", "default": False}, "jump_monitor": {"type": "boolean", "description": "Enable semi-auto JUMP monitoring. Default false.", "default": False}, "jump_monitor_interval": {"type": "integer", "description": "How often (in seconds) to check jump conditions. Default 10.", "default": 10}, }, "required": ["markets"] } }, { "name": "schedule_session", "description": "Schedule a trading session for the future. Provide start/stop times as relative (+30m, +1h) or absolute (HH:MM UTC, ISO-8601).", "input_schema": { "type": "object", "properties": { "markets": {"type": "array", "items": {"type": "string"}, "description": "Market tickers"}, "start_time": {"type": "string", "description": "When to start: '+30m', '+1h', '15:30', or ISO-8601"}, "stop_time": {"type": "string", "description": "When to gracefully stop (optional)"}, "contract_size": {"type": "integer", "default": 1}, "jump_instructions": {"type": "string", "description": "Natural language description of when to jump"}, }, "required": ["markets", "start_time"] } }, { "name": "control_instance", "description": "Send a control command to the active trading instance.", "input_schema": { "type": "object", "properties": { "action": { "type": "string", "enum": ["cancel_orders", "toggle_pause", "single_fire", "stop", "force_stop"], "description": "The control action. cancel_orders = cancel all resting orders immediately (z key). stop = graceful stop after cycle. force_stop = cancel orders + stop instance.", }, "instance_id": {"type": "integer", "description": "Instance ID (uses active if omitted)"}, }, "required": ["action"] } }, { "name": "toggle_jump", "description": "Toggle JUMP mode on a market. JUMP pennies ahead by 1 cent. market_number is 1 or 2 (human-readable, not 0-indexed).", "input_schema": { "type": "object", "properties": { "market_number": {"type": "integer", "description": "1 = first market, 2 = second market"}, "instance_id": {"type": "integer", "description": "Instance ID (uses active if omitted)"}, }, "required": ["market_number"] } }, { "name": "get_status", "description": "Get status of all running trading instances (positions, fills, P&L).", "input_schema": {"type": "object", "properties": {}} }, { "name": "get_balance", "description": "Check Base wallet USDC balance.", "input_schema": {"type": "object", "properties": {}} }, { "name": "list_scheduled_sessions", "description": "List all pending/running scheduled trading sessions.", "input_schema": {"type": "object", "properties": {}} }, { "name": "cancel_session", "description": "Cancel a pending scheduled trading session.", "input_schema": { "type": "object", "properties": { "session_id": {"type": "string", "description": "Session ID (first 8 chars is enough)"}, }, "required": ["session_id"] } }, { "name": "list_reports", "description": "List on-chain trading session reports by author Base address.", "input_schema": { "type": "object", "properties": { "author_address": {"type": "string", "description": "Base EVM address of the report author"}, }, "required": ["author_address"] } }, { "name": "read_report", "description": "Fetch and decrypt an on-chain trading session report from Base. Requires pledged syndicate membership.", "input_schema": { "type": "object", "properties": { "tx_hash": {"type": "string", "description": "Transaction hash of the report on Base"}, "author_address": {"type": "string", "description": "Base EVM address of the report author"}, "requester_address": {"type": "string", "description": "Your Base EVM address (must be pledged)"}, }, "required": ["tx_hash", "author_address", "requester_address"] } }, { "name": "schedule_games", "description": "Schedule a LuckBot Trader to trade a list of specific games in the next 24 hours. Each game needs market tickers and a start time. The schedule is committed to Base on-chain and auto-deployed at game time.", "input_schema": { "type": "object", "properties": { "games": { "type": "array", "items": { "type": "object", "properties": { "markets": {"type": "array", "items": {"type": "string"}, "description": "1 or 2 Kalshi market tickers"}, "start_time": {"type": "string", "description": "When to deploy: ISO-8601 UTC (e.g. '2026-02-21T20:00:00Z') or relative ('+30m', '+2h')"}, "stop_time": {"type": "string", "description": "When to stop (optional, same format)"}, "contract_size": {"type": "integer", "default": 1}, "higher_first": {"type": "boolean", "default": False, "description": "Use higher-first mode (recommended for basketball)"}, }, "required": ["markets", "start_time"] }, "description": "List of games to schedule" }, }, "required": ["games"] } }, { "name": "list_scheduled_games", "description": "List all pending and active scheduled games.", "input_schema": {"type": "object", "properties": {}} }, ] # ============================================================================ # SYSTEM PROMPT # ============================================================================ OPENCLAWD_SYSTEM = """You are OpenClawd, a LuckySt Agentic Market Making bot on Kalshi prediction markets. You have TOOLS to deploy, control, and schedule trading instances. Use them when the operator gives you trading instructions. ## When to use tools vs just talk: - If the operator gives a trading instruction (deploy, jump, stop, schedule), use tools. - If they ask a question about strategy, markets, or how things work, just answer conversationally. - If information is missing (market tickers, timing, size), ask before calling tools. ## Tools available: - deploy_instance: Deploy NOW on 1-2 markets - schedule_session: Schedule future trading with start/stop times - control_instance: cancel_orders (z key - cancel all orders NOW), toggle_pause, single_fire, stop (graceful - waits for cycle), force_stop (cancel orders + stop instance) - toggle_jump: Penny ahead 1c on a market (market_number: 1 or 2) - get_status: Check running instances - get_balance: Check Base wallet USDC - list_reports / read_report: View on-chain session reports from Base - schedule_games: Schedule games for the next 24h (committed to Base on-chain) - list_scheduled_games: See pending & active scheduled games - list_scheduled_sessions / cancel_session: Manage scheduled sessions ## JOIN Mode (Continuous) JOIN is the default mode. The agent places resting limit orders at the best bid on both sides of a binary market, profiting the spread when both fill. Price selection logic: - If others have larger size at best bid -> join them at that price - If we're the best bid -> fall back to 2nd best bid to avoid being alone at top - Always join the queue, never cross the spread ## JUMP (Opt-in Semi-Auto + Manual) JUMP pennies ahead by 1 cent on a specific market. **Semi-auto monitor (opt-in):** Set jump_monitor=true on deploy to enable. The Luckbot checks conditions every N seconds (default 10, configurable via jump_monitor_interval) and triggers JUMP when ALL conditions are met: 1. Spread between yes/no bids > 1c in that market 2. Our queue position is in the last 25% of total contracts at that price 3. Our resting order is < 25% of total contracts at that price level The monitor is OFF by default. Only enable it if the operator asks for it. **Manual:** You can trigger JUMP anytime via toggle_jump tool (market_number 1 or 2). How JUMP works: 1. A penny-ahead target is set: others_best_bid + 1 cent 2. If others match or exceed the jump price, jump auto-disables 3. Agent returns to JOIN on that market When NOT to JUMP: - Spread would collapse below 2c after jumping - Already at or near the best bid with good queue position - Market is volatile (rapid price changes waste the jump) - Both sides filling steadily in JOIN mode ## NO-Bid Market Making (2-market) On binary events (Team A vs Team B), bid NO on both sides. - Spread = 100c - NO_bid_A_cents - NO_bid_B_cents - If both NO bids fill, profit = spread per contract ## HIGHER-FIRST Mode (2-market) Higher-first trades one market at a time, starting with whichever has the HIGHER NO bid. - The higher NO bid = the team more likely to win - Filling that side first = positive time decay WHEN TO USE higher_first=true: - Basketball (NBA, NCAAMB): STRONGLY RECOMMENDED - Hockey (NHL): Useful - Esports (match-winner BO3/BO5): Useful - Crypto intervals: Optional WHEN NOT TO USE: - 50/50 markets - When you want max fill speed - Football/soccer moneylines DEFAULT BEHAVIOR: For basketball, always use higher_first=true unless the operator says otherwise. ## Markets to AVOID - Football (NFL, NCAAF) moneylines: Touchdowns cause 20-30c jumps - Soccer moneylines: Goals cause 30-40c jumps ## Risk Rules: 1. Never chase -- if spread collapsed below 2c, let it go 2. Respect position limits -- hard stops, no exceptions 3. Pause during known events -- breaking news, major in-game events 4. Monitor fill imbalance -- JUMP to resolve stuck positions 5. Reduce size near resolution -- prices become binary ## Style: - Concise. You're a trader. - Use market slang: fills, spreads, queue, penny, edge. - End key messages with: Syndicate eternal. Spreads temporary. Edge ours. Current trading context will be provided when available.""" # ============================================================================ # TOOL EXECUTION # ============================================================================ async def execute_tool( tool_name: str, tool_input: dict, chat_id: Optional[int] = None, jump_monitor_fn: Optional[Callable] = None, stop_monitor_fn: Optional[Callable] = None, schedule_fn: Optional[Callable] = None, sessions: Optional[dict] = None, ) -> str: """ Execute a tool call and return result string for Claude. Adapter-specific callbacks: - jump_monitor_fn(chat_id, interval): start jump monitor (Telegram only) - stop_monitor_fn(): stop jump monitor (Telegram only) - schedule_fn(session_data): schedule a future session (Telegram only) - sessions: dict of active scheduled sessions (Telegram only) """ token, api_key, rsa_key = get_token_and_creds() if tool_name == "deploy_instance": if not token or not api_key or not rsa_key: return "Error: Not logged in or missing Kalshi credentials. Use /login and /creds first." markets = tool_input["markets"] size = tool_input.get("contract_size", 1) higher_first = tool_input.get("higher_first", False) jump_monitor = tool_input.get("jump_monitor", False) jump_interval = tool_input.get("jump_monitor_interval", 10) num = len(markets) market_priority = "expensive" if (higher_first and num == 2) else ("none" if num == 2 else None) config = { "num_markets": num, "mode": "automated", "markets": markets, "kalshi_api_key": api_key, "rsa_key_path": rsa_key, "both_side": "no" if num == 2 else None, "market_priority": market_priority, "side_priority": None if num == 2 else "no", "min_spread": 2, "max_spread": 15, "m1_bounds": [1, 15, 85, 99], "m2_bounds": [1, 15, 85, 99] if num == 2 else None, "position_increment": size, "max_position": 100, "join_only": True, "grid_mode": False, "grid_levels": None, "contract_increment": size, } if chat_id: config["telegram_chat_id"] = str(chat_id) result = await api_request("POST", "/terminal/auto/deploy", config, token) if "error" in result: return f"Deploy failed: {result['error']}" instance_id = result.get("id") session = load_session() session["active_instance"] = instance_id save_session(session) monitor_str = "" if jump_monitor and jump_monitor_fn and chat_id: jump_monitor_fn(chat_id, jump_interval) monitor_str = f" | Jump monitor ON ({jump_interval}s)" elif jump_monitor: monitor_str = " | Jump monitor requested (not available in this interface)" market_str = " vs ".join(m.split("-")[-1] for m in markets) mode_str = "JOIN + HIGHER-FIRST" if higher_first else "JOIN" return f"Deployed instance #{instance_id}. {market_str} | {size} contracts | {mode_str} mode.{monitor_str}" elif tool_name == "schedule_session": if not schedule_fn: return "Scheduling is not available in this interface. Deploy immediately with deploy_instance instead." if not token or not api_key or not rsa_key: return "Error: Not logged in or missing credentials." return await schedule_fn(tool_input, chat_id) elif tool_name == "control_instance": if not token: return "Error: Not logged in." instance_id = tool_input.get("instance_id") or get_active_instance() if not instance_id: return "Error: No active instance." action = tool_input["action"] result = await api_request("POST", f"/terminal/auto/instances/{instance_id}/control", {"action": action}, token) if "error" in result: return f"Error: {result['error']}" if action in ("stop", "force_stop") and stop_monitor_fn: stop_monitor_fn() return f"Instance #{instance_id}: {action} sent." elif tool_name == "toggle_jump": if not token: return "Error: Not logged in." instance_id = tool_input.get("instance_id") or get_active_instance() if not instance_id: return "Error: No active instance." market_index = tool_input["market_number"] - 1 result = await api_request("POST", f"/terminal/auto/instances/{instance_id}/jump", {"market_index": market_index}, token) if "error" in result: return f"Error: {result['error']}" active = result.get("jump_active", False) return f"Market {tool_input['market_number']} JUMP: {'ON' if active else 'OFF'}" elif tool_name == "get_status": if not token: return "Not logged in." result = await api_request("GET", "/terminal/auto/instances", token=token) if "error" in result: return f"Error: {result['error']}" instances = result.get("instances", []) if not instances: return "No active instances." lines = [] for inst in instances: markets = inst.get("markets", []) market_str = " vs ".join(m.split("-")[-1] for m in markets) inc = inst.get("current_increment", {}) fill_str = "" if "m1" in inc: fill_str = f" | Fills: M1 {inc['m1'].get('filled', 0)}/{inc['m1'].get('total', 0)}, M2 {inc['m2'].get('filled', 0)}/{inc['m2'].get('total', 0)}" elif "yes" in inc: fill_str = f" | Fills: Y {inc['yes'].get('filled', 0)}/{inc['yes'].get('total', 0)}, N {inc['no'].get('filled', 0)}/{inc['no'].get('total', 0)}" lines.append( f"#{inst['id']} {inst.get('status', '?').upper()} | {market_str} | " f"Pos: {inst.get('position', 0)} | P&L: {inst.get('pnl', '$0.00')}{fill_str}" ) return "\n".join(lines) elif tool_name == "get_balance": wallet = os.environ.get("LUCKBOT_BASE_ADDRESS", "0x4085b0eA2598aF729205bC982b2A3419593BE24d") network = os.environ.get("LUCKYST_NETWORK", "mainnet") rpc_url = "https://mainnet.base.org" if network == "mainnet" else "https://sepolia.base.org" # USDC on Base (ERC-20): mainnet 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913 usdc_addr = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" if network == "mainnet" else "0x036CbD53842c5426634e7929541eC2318f3dCF7e" try: selector = "0x70a08231" padded = wallet.lower().replace("0x", "").zfill(64) async with httpx.AsyncClient() as client: resp = await client.post(rpc_url, json={ "jsonrpc": "2.0", "method": "eth_call", "params": [{"to": usdc_addr, "data": selector + padded}, "latest"], "id": 1, }) usdc = int(resp.json()["result"], 16) / 1e6 return f"Base Wallet: {wallet[:8]}...{wallet[-6:]}\nUSDC: {usdc:.2f}" except Exception as e: return f"Error checking balance: {e}" elif tool_name == "list_reports": author = tool_input["author_address"] result = await api_request("GET", f"/syndicate/reports/by-author/{author}", token=token) if "error" in result: return f"Error: {result['error']}" if isinstance(result, list) and not result: return f"No reports found for {author[:10]}..." lines = [] for rpt in result: markets_str = ", ".join(rpt.get("markets", [])) lines.append(f"{rpt['tx_hash'][:16]}... | Instance #{rpt.get('instance_id', '?')} | {markets_str}") return "\n".join(lines) elif tool_name == "read_report": result = await api_request("POST", "/syndicate/reports/decrypt", { "tx_hash": tool_input["tx_hash"], "requester_address": tool_input["requester_address"], "author_address": tool_input["author_address"], }, token=token) if "error" in result: return f"Error: {result['error']}" return result.get("plaintext", "No data returned.") elif tool_name == "list_scheduled_sessions": if not sessions: return "No scheduled sessions." lines = [] for sid, sess in sessions.items(): market_str = " vs ".join(m.split("-")[-1] for m in sess.markets) start_str = sess.start_time.strftime("%H:%M UTC") if sess.start_time else "?" stop_str = sess.stop_time.strftime("%H:%M UTC") if sess.stop_time else "manual" inst_str = f" | Instance #{sess.instance_id}" if sess.instance_id else "" lines.append(f"{sid[:8]} | {sess.status.upper()} | {market_str} | Start: {start_str} | Stop: {stop_str}{inst_str}") return "\n".join(lines) elif tool_name == "cancel_session": if not sessions: return "No sessions to cancel." target = tool_input["session_id"] match = None for sid in sessions: if sid.startswith(target) or sid[:8] == target: match = sid break if not match: return f"Session {target} not found." sessions[match].status = "cancelled" return f"Session {match[:8]} cancelled." elif tool_name == "schedule_games": games = tool_input["games"] # Parse relative times (e.g. "+30m", "+2h") into ISO-8601 from datetime import datetime as dt, timezone as tz, timedelta now = dt.now(tz.utc) for g in games: st = g.get("start_time", "") if st.startswith("+"): g["start_time"] = _parse_relative_time(st, now).isoformat() stop = g.get("stop_time") if stop and stop.startswith("+"): g["stop_time"] = _parse_relative_time(stop, now).isoformat() result = await api_request("POST", "/syndicate/schedule/games", {"games": games}, token=token) if "error" in result: return f"Schedule failed: {result['error']}" sid = result.get("schedule_id", "?") count = result.get("game_count", 0) tx = result.get("base_tx") tx_str = f"\nBase commitment: {tx}" if tx else "" return f"Scheduled {count} game(s). ID: {sid}{tx_str}" elif tool_name == "list_scheduled_games": pending = await api_request("GET", "/syndicate/schedule/pending", token=token) active = await api_request("GET", "/syndicate/schedule/active", token=token) lines = [] if isinstance(pending, list) and pending: lines.append("PENDING:") for g in pending: markets_str = " vs ".join(m.split("-")[-1] for m in g.get("markets", [])) lines.append(f" {g.get('schedule_id', '?')}:{g.get('game_index', 0)} | {markets_str} | Start: {g.get('start_time', '?')}") if isinstance(active, list) and active: lines.append("ACTIVE:") for g in active: markets_str = " vs ".join(m.split("-")[-1] for m in g.get("markets", [])) lines.append(f" {g.get('schedule_id', '?')}:{g.get('game_index', 0)} | {markets_str} | Instance #{g.get('instance_id', '?')}") return "\n".join(lines) if lines else "No scheduled games." return f"Unknown tool: {tool_name}" def _parse_relative_time(s: str, now) -> "datetime": """Parse '+30m', '+2h', '+1h30m' into a datetime.""" from datetime import timedelta s = s.lstrip("+").strip() total_minutes = 0 if "h" in s: parts = s.split("h") total_minutes += int(parts[0]) * 60 s = parts[1] if len(parts) > 1 else "" if "m" in s: s = s.replace("m", "").strip() if s: total_minutes += int(s) return now + timedelta(minutes=total_minutes) # ============================================================================ # CLAUDE AGENTIC LOOP # ============================================================================ def _content_to_dicts(content_blocks) -> list: """Convert pydantic content blocks to plain dicts for serialization.""" result = [] for block in content_blocks: if block.type == "text": result.append({"type": "text", "text": block.text}) elif block.type == "tool_use": result.append({"type": "tool_use", "id": block.id, "name": block.name, "input": block.input}) return result def _sanitize_history(history: list) -> list: """Clean history to avoid orphaned tool_use/tool_result pairs.""" if len(history) > 20: history = history[-20:] while history and history[0].get("role") == "assistant": history.pop(0) while history and history[0].get("role") == "user": content = history[0].get("content", "") if isinstance(content, list): history.pop(0) while history and history[0].get("role") == "assistant": history.pop(0) else: break return history async def run_claude_agentic_loop( user_message: str, chat_history: list, tool_executor: Callable[[str, dict], Awaitable[str]], system_prompt_extra: str = "", system_prompt_override: str = "", max_iterations: int = 5, ) -> tuple[str, list]: """ Run the Claude tool-use agentic loop. Args: user_message: The user's text message. chat_history: Conversation history (modified in place and returned). tool_executor: async fn(tool_name, tool_input) -> result_str system_prompt_extra: Additional context appended to system prompt. system_prompt_override: If set, replaces OPENCLAWD_SYSTEM entirely (for role-based prompts). max_iterations: Max tool-use iterations. Returns: (reply_text, updated_history) """ if not HAS_ANTHROPIC or not ANTHROPIC_API_KEY: return ("AI chat not configured. Set ANTHROPIC_API_KEY.", chat_history) history = list(chat_history) history.append({"role": "user", "content": user_message}) history = _sanitize_history(history) base_prompt = system_prompt_override if system_prompt_override else OPENCLAWD_SYSTEM system = base_prompt + system_prompt_extra client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY) try: response = None for _ in range(max_iterations): response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, system=system, tools=TRADING_TOOLS, messages=history, ) content_dicts = _content_to_dicts(response.content) history.append({"role": "assistant", "content": content_dicts}) if response.stop_reason == "tool_use": tool_results = [] for block in response.content: if block.type == "tool_use": logger.info(f"Tool call: {block.name}({json.dumps(block.input)})") try: result_str = await tool_executor(block.name, block.input) except Exception as tool_err: logger.error(f"Tool {block.name} error: {tool_err}") result_str = f"Error executing {block.name}: {tool_err}" tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": result_str, }) history.append({"role": "user", "content": tool_results}) else: break # Extract final text reply_parts = [] if response: for block in response.content: if hasattr(block, "text"): reply_parts.append(block.text) reply = "\n".join(reply_parts) if reply_parts else "Done." return (reply, history) except Exception as e: logger.error(f"Claude agentic loop error: {e}") return (f"AI error: {e}", [])