import json import time from datetime import UTC, datetime from typing import Any, cast import anthropic import structlog from marketing_intelligence.agent.mcp_client import MCPToolClient from marketing_intelligence.core.config import settings logger = structlog.get_logger("agent.anthropic") def _key_inputs(name: str, inputs: dict[str, Any]) -> dict[str, Any]: if name in ("scrape_sound_page", "build_sound_url"): return {"sound_id": inputs.get("sound_id")} if name in ("scrape_video", "scrape_url"): return {"url": inputs.get("url")} if name == "scrape_tag_page": return {"tag": inputs.get("tag")} if name == "export_results": return {"label": inputs.get("label")} if name in ("close_browser_session", "apply_behavior_tactics"): return {"session_id": inputs.get("session_id")} return {} def _key_result(name: str, result: Any) -> dict[str, Any]: if not isinstance(result, dict): return {} ok = result.get("success", True) base = {"ok": ok} if not ok: return {**base, "error": str(result.get("error", ""))[:120]} if name == "scrape_tag_page": return { **base, "videos": len(result.get("videos", [])), "video_count_text": result.get("video_count_text"), } if name == "scrape_sound_page": return { **base, "title": result.get("title"), "video_urls": len(result.get("video_urls", [])), "video_count_text": result.get("video_count_text"), } if name == "scrape_video": return { **base, "views": result.get("views"), "likes": result.get("likes"), "sound_id": result.get("sound_id"), } if name == "scrape_url": return {**base, "chars": len(str(result.get("content", "")))} if name == "export_results": return {**base, "path": result.get("saved_to"), "s3_uri": result.get("s3_uri")} block = (result.get("signals") or {}).get("block_signal") if block and block not in ("none", None): base["block_signal"] = block return base def _prune_old_results(messages: list[dict[str, Any]], keep_last: int = 6) -> None: result_items = [ item for msg in messages if msg["role"] == "user" and isinstance(msg["content"], list) for item in msg["content"] if isinstance(item, dict) and item.get("type") == "tool_result" ] for item in result_items[:-keep_last]: item["content"] = "ok" class AnthropicAgent: def __init__(self, model_id: str | None = None) -> None: self._client = anthropic.Anthropic(api_key=settings.anthropic_api_key) self._model_id = model_id or settings.anthropic_model_id async def run( self, task: str, params: dict[str, Any] | None = None, run_id: str | None = None, allowed_tools: set[str] | None = None, ) -> str: messages: list[dict[str, Any]] = [{"role": "user", "content": task}] step = 0 agent_start = time.time() run_id = run_id or datetime.now(UTC).strftime("%Y%m%d_%H%M%S") # bound_contextvars — not just logger.bind() — so track_name/campaign_id also reach # deep MCP-tool-internal logs via the merge_contextvars processor, without threading # them through every tool signature. Scoped to this call: each Stage 1 track runs in # its own asyncio Task with its own context copy, so concurrent tracks never see each # other's bound vars. log_context = { k: v for k, v in (params or {}).items() if k in ("track_name", "campaign_id") } with structlog.contextvars.bound_contextvars(**log_context): log = logger.bind(model=self._model_id, backend="anthropic") log.info("agent.start", task=task, params=params or {}) async with MCPToolClient() as mcp: tool_specs = await mcp.list_tools_anthropic(allowed=allowed_tools) if tool_specs: tool_specs = tool_specs[:-1] + [ {**tool_specs[-1], "cache_control": {"type": "ephemeral"}} ] system_prompt = await mcp.get_prompt("tiktok_system", params or {}) while True: _prune_old_results(messages) with self._client.messages.stream( model=self._model_id, max_tokens=32000, system=[ { "type": "text", "text": system_prompt, "cache_control": {"type": "ephemeral"}, } ], tools=tool_specs, # type: ignore[arg-type] messages=messages, # type: ignore[arg-type] ) as stream: response = stream.get_final_message() log.info( "agent.turn.usage", input_tokens=response.usage.input_tokens, output_tokens=response.usage.output_tokens, cache_creation_input_tokens=response.usage.cache_creation_input_tokens, cache_read_input_tokens=response.usage.cache_read_input_tokens, ) messages.append({"role": "assistant", "content": response.content}) if response.stop_reason == "end_turn": elapsed_ms = int((time.time() - agent_start) * 1000) log.info("agent.done", steps=step, elapsed_ms=elapsed_ms) text_blocks = [ b.text for b in response.content if hasattr(b, "text") ] return text_blocks[0] if text_blocks else "" if response.stop_reason != "tool_use": return f"Unexpected stop reason: {response.stop_reason}" tool_results = [] for block in response.content: if block.type != "tool_use": continue step += 1 inputs = cast(dict[str, Any], block.input) key_in = _key_inputs(block.name, inputs) log.info("tool.call", step=step, tool=block.name, **key_in) t0 = time.time() if ( allowed_tools is not None and block.name not in allowed_tools ): log.warning( "tool.rejected", step=step, tool=block.name, reason="not in allowed_tools", ) result = { "error": f"tool '{block.name}' is not available in this stage", "success": False, } else: try: result = await mcp.call_tool(block.name, inputs) except Exception as exc: result = {"error": str(exc), "success": False} elapsed_ms = int((time.time() - t0) * 1000) key_out = _key_result(block.name, result) log.info( "tool.result", step=step, tool=block.name, elapsed_ms=elapsed_ms, **key_out, ) tool_results.append( { "type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result, default=str), } ) messages.append({"role": "user", "content": tool_results})