diff --git a/PROXY-API-TODO.md b/PROXY-API-TODO.md index 03fc06c..0b6cc60 100644 --- a/PROXY-API-TODO.md +++ b/PROXY-API-TODO.md @@ -4,11 +4,11 @@ Future work for Vimini's proxy/API backend support. ## Anthropic API support -- Add a native Anthropic backend in addition to the current OpenAI-compatible proxy backend. -- Map Vimini's generic chat, streaming, tools, and tool-response abstractions to Anthropic's Messages API. -- Handle Anthropic-specific streaming events and content block formats. -- Add global config support for `api: "anthropic"`, including model defaults and API key file loading. -- Add tests for model listing behavior, streaming responses, tool calls, and error handling. +- [x] Add a native Anthropic backend in addition to the current OpenAI-compatible proxy backend. +- [x] Map Vimini's generic chat, streaming, tools, and tool-response abstractions to Anthropic's Messages API. +- [x] Handle Anthropic-specific streaming events and content block formats. +- [x] Add global config support for `api: "anthropic"`, including model defaults and API key file loading. +- [x] Add tests for model listing behavior, streaming responses, tool calls, and error handling. ## OpenAI Files API for context files diff --git a/README.md b/README.md index adef2bd..f040664 100644 --- a/README.md +++ b/README.md @@ -179,7 +179,7 @@ customizations. } ``` - You can configure multiple backends (Google Gemini via `"api": "genai"`, or OpenAI-compatible proxies and endpoints via `"api": "openai"`). + You can configure multiple backends (Google Gemini via `"api": "genai"`, OpenAI-compatible proxies and endpoints via `"api": "openai"`, or Anthropic via `"api": "anthropic"`). Backends that require credentials specify an `api_key_file`; Gemini requires one, while OpenAI-compatible local/proxy endpoints may omit it unless @@ -194,7 +194,7 @@ Opens an interactive configuration editor in a new split window to configure glo **How to use the global configuration manager:** * Move the cursor to any configuration option or backend line. * **`` or `e`**: Edits the selected value. On `active_backend`, enter one of the configured backend names. On a backend header, rename the backend. Under `UI PREFERENCES`, customize colors (e.g. `thinking_color`, `prompt_color`, `waiting_color`, `service_color`, `error_color`, `prompt_hint_color`). -* **`a`**: Adds a new backend and prompts for API type, default model, API key file, and base URL for OpenAI-compatible backends. +* **`a`**: Adds a new backend and prompts for API type, default model, API key file, and base URL for OpenAI-compatible or Anthropic backends. * **`d`**: Clears the selected backend field, deletes a backend after confirmation, or resets a UI preference back to default. The last backend cannot be deleted. * **`r`**: Resets all global configuration options to their last saved state. * **`q`**: Closes the configuration window. If modifications were made, a confirmation prompt asks whether to save changes permanently to `~/.var/vimini/config.json`. diff --git a/python3/vimini/common/backends/__init__.py b/python3/vimini/common/backends/__init__.py index c6838e2..ad9ae0d 100644 --- a/python3/vimini/common/backends/__init__.py +++ b/python3/vimini/common/backends/__init__.py @@ -18,6 +18,12 @@ convert_schema_to_openai, convert_tools_to_openai, ) +from vimini.common.backends.anthropic_backend import ( + AnthropicBackend, + AnthropicChatSession, + convert_schema_to_anthropic, + convert_tools_to_anthropic, +) from vimini.common.config import ( load_global_config, get_active_backend, @@ -85,6 +91,13 @@ def get_backend( base_url=backend_cfg.get("base_url"), headers=backend_cfg.get("headers"), ) + elif api == "anthropic": + return AnthropicBackend( + config=backend_cfg, + api_key=api_key, + base_url=backend_cfg.get("base_url"), + headers=backend_cfg.get("headers"), + ) raise ValueError(f"Unsupported API protocol '{api}' for backend '{backend_name}'.") @@ -123,8 +136,12 @@ def get_client( "GenAIChatSession", "ProxyBackend", "ProxyChatSession", + "AnthropicBackend", + "AnthropicChatSession", "convert_schema_to_openai", "convert_tools_to_openai", + "convert_schema_to_anthropic", + "convert_tools_to_anthropic", "get_backend", "get_client", ] diff --git a/python3/vimini/common/backends/anthropic_backend.py b/python3/vimini/common/backends/anthropic_backend.py new file mode 100644 index 0000000..02567a4 --- /dev/null +++ b/python3/vimini/common/backends/anthropic_backend.py @@ -0,0 +1,618 @@ +import json +import logging +import urllib.request +import urllib.error +from typing import Any, Dict, Iterator, List, Optional + +from vimini.common.backends.base import ( + BaseChatSession, + BaseClient, + StreamChunk, + ToolCall, + ToolResponse, +) +from vimini.common.config import load_backend_api_key + +logger = logging.getLogger("vimini_agent") + + +def convert_schema_to_anthropic(schema: Any) -> Dict[str, Any]: + """ + Converts a Google GenAI types.Schema, Vimini Schema, or dict into + an Anthropic-compatible JSON schema. + """ + if schema is None: + return {"type": "object", "properties": {}} + + if isinstance(schema, dict): + res = dict(schema) + if "type" in res: + res["type"] = str(res["type"]).lower().split(".")[-1] + if "properties" in res and isinstance(res["properties"], dict): + res["properties"] = { + k: convert_schema_to_anthropic(v) for k, v in res["properties"].items() + } + if "items" in res and res["items"] is not None: + res["items"] = convert_schema_to_anthropic(res["items"]) + res.setdefault("type", "object") + if res.get("type") == "object": + res.setdefault("properties", {}) + return res + + type_val = getattr(schema, "type", "object") + type_str = str(type_val).lower().split(".")[-1] + if type_str == "type": + type_str = "object" + + res: Dict[str, Any] = {"type": type_str} + + desc = getattr(schema, "description", None) + if desc: + res["description"] = desc + + properties = getattr(schema, "properties", None) + if properties and isinstance(properties, dict): + res["properties"] = { + k: convert_schema_to_anthropic(v) for k, v in properties.items() + } + + required = getattr(schema, "required", None) + if required and isinstance(required, list): + res["required"] = required + + enum = getattr(schema, "enum", None) + if enum and isinstance(enum, list): + res["enum"] = enum + + items = getattr(schema, "items", None) + if items is not None: + res["items"] = convert_schema_to_anthropic(items) + + if res.get("type") == "object": + res.setdefault("properties", {}) + + return res + + +def convert_tools_to_anthropic(tools: Any) -> List[Dict[str, Any]]: + """ + Converts a list of GenAI Tool, FunctionDeclaration, or dict objects + to Anthropic tools format. + """ + if not tools: + return [] + + anthropic_tools: List[Dict[str, Any]] = [] + tool_items = tools if isinstance(tools, (list, tuple)) else [tools] + + for item in tool_items: + if isinstance(item, dict): + if "name" in item and "input_schema" in item: + anthropic_tools.append(item) + elif item.get("type") == "function" and "function" in item: + fn = item["function"] + anthropic_tools.append( + { + "name": fn.get("name", ""), + "description": fn.get("description", ""), + "input_schema": convert_schema_to_anthropic( + fn.get("parameters") + ), + } + ) + elif "name" in item: + anthropic_tools.append( + { + "name": item["name"], + "description": item.get("description", ""), + "input_schema": convert_schema_to_anthropic( + item.get("parameters") or item.get("input_schema") + ), + } + ) + continue + + # Handle google.genai.types.Tool or vimini.common.backends.base.Tool + if hasattr(item, "function_declarations") and item.function_declarations: + for fd in item.function_declarations: + anthropic_tools.append( + { + "name": fd.name, + "description": getattr(fd, "description", "") or "", + "input_schema": convert_schema_to_anthropic( + getattr(fd, "parameters", None) + ), + } + ) + continue + + # Handle FunctionDeclaration + if hasattr(item, "name"): + anthropic_tools.append( + { + "name": item.name, + "description": getattr(item, "description", "") or "", + "input_schema": convert_schema_to_anthropic( + getattr(item, "parameters", None) + ), + } + ) + + return anthropic_tools + + +class AnthropicChatSession(BaseChatSession): + def __init__( + self, + backend: "AnthropicBackend", + model: str, + system_instruction: Optional[str] = None, + tools: Optional[List[Dict[str, Any]]] = None, + ): + self.backend = backend + self.model = model + self.system_instruction = system_instruction + self.tools = tools or [] + self.messages: List[Dict[str, Any]] = [] + + def has_history(self) -> bool: + return False + + def get_history(self) -> List[Any]: + return [] + + def _append_user_content(self, content: Any) -> None: + """ + Appends user content while strictly preserving Anthropic's rule that + messages must alternate between 'user' and 'assistant'. + """ + if self.messages and self.messages[-1].get("role") == "user": + prev = self.messages[-1]["content"] + if isinstance(prev, str) and isinstance(content, str): + self.messages[-1]["content"] = f"{prev}\n\n{content}" + elif isinstance(prev, list) and isinstance(content, list): + self.messages[-1]["content"] = prev + content + elif isinstance(prev, list) and isinstance(content, str): + self.messages[-1]["content"] = prev + [ + {"type": "text", "text": content} + ] + elif isinstance(prev, str) and isinstance(content, list): + self.messages[-1]["content"] = [ + {"type": "text", "text": prev} + ] + content + else: + self.messages.append({"role": "user", "content": content}) + + def send_message_stream( + self, prompt_or_tool_responses: Any + ) -> Iterator[StreamChunk]: + if isinstance(prompt_or_tool_responses, str): + self._append_user_content(prompt_or_tool_responses) + elif isinstance(prompt_or_tool_responses, ToolResponse): + content = ( + prompt_or_tool_responses.response + if isinstance(prompt_or_tool_responses.response, str) + else json.dumps(prompt_or_tool_responses.response) + ) + self._append_user_content( + [ + { + "type": "tool_result", + "tool_use_id": prompt_or_tool_responses.call_id or "", + "content": content, + } + ] + ) + elif isinstance(prompt_or_tool_responses, list): + tool_results = [] + text_items = [] + for item in prompt_or_tool_responses: + if isinstance(item, ToolResponse): + content = ( + item.response + if isinstance(item.response, str) + else json.dumps(item.response) + ) + tool_results.append( + { + "type": "tool_result", + "tool_use_id": item.call_id or "", + "content": content, + } + ) + elif isinstance(item, dict): + if item.get("type") == "tool_result": + tool_results.append(item) + elif "tool_call_id" in item: + tool_results.append( + { + "type": "tool_result", + "tool_use_id": item["tool_call_id"], + "content": str(item.get("content", "")), + } + ) + elif "role" in item: + self.messages.append(item) + else: + tool_results.append(item) + else: + text_items.append(str(item)) + + if tool_results: + self._append_user_content(tool_results) + elif text_items: + self._append_user_content("\n\n".join(text_items)) + elif isinstance(prompt_or_tool_responses, dict): + if "role" in prompt_or_tool_responses: + self.messages.append(prompt_or_tool_responses) + else: + self._append_user_content(json.dumps(prompt_or_tool_responses)) + + payload: Dict[str, Any] = { + "model": self.model, + "max_tokens": self.backend.max_tokens, + "messages": list(self.messages), + "stream": True, + } + if self.system_instruction: + payload["system"] = self.system_instruction + if self.tools: + payload["tools"] = self.tools + + resp = self.backend._make_request("/messages", data=payload, stream=True) + + accumulated_text = "" + accumulated_thought = "" + content_blocks: Dict[int, Dict[str, Any]] = {} + + for event_data in self.backend._read_sse_stream(resp): + event_type = event_data.get("type") + + if event_type == "content_block_start": + idx = event_data.get("index", 0) + content_blocks[idx] = dict(event_data.get("content_block", {})) + if content_blocks[idx].get("type") == "tool_use": + content_blocks[idx]["partial_json"] = "" + + elif event_type == "content_block_delta": + idx = event_data.get("index", 0) + delta = event_data.get("delta", {}) + delta_type = delta.get("type") + text_chunk = "" + thought_chunk = "" + + if delta_type == "text_delta": + text_chunk = delta.get("text", "") or "" + accumulated_text += text_chunk + if idx not in content_blocks: + content_blocks[idx] = {"type": "text", "text": ""} + content_blocks[idx]["text"] = ( + content_blocks[idx].get("text", "") + text_chunk + ) + elif delta_type == "thinking_delta": + thought_chunk = delta.get("thinking", "") or "" + accumulated_thought += thought_chunk + if idx not in content_blocks: + content_blocks[idx] = {"type": "thinking", "thinking": ""} + content_blocks[idx]["thinking"] = ( + content_blocks[idx].get("thinking", "") + thought_chunk + ) + elif delta_type == "signature_delta": + sig = delta.get("signature", "") or "" + if idx in content_blocks: + content_blocks[idx]["signature"] = ( + content_blocks[idx].get("signature", "") + sig + ) + elif delta_type == "input_json_delta": + partial = delta.get("partial_json", "") or "" + if idx not in content_blocks: + content_blocks[idx] = {"type": "tool_use", "partial_json": ""} + content_blocks[idx]["partial_json"] = ( + content_blocks[idx].get("partial_json", "") + partial + ) + + if text_chunk or thought_chunk: + yield StreamChunk(text=text_chunk, thought=thought_chunk) + + elif event_type == "content_block_stop": + pass + + # Assemble completed tool calls + completed_tool_calls: List[ToolCall] = [] + for idx in sorted(content_blocks.keys()): + block = content_blocks[idx] + if block.get("type") == "tool_use": + call_id = block.get("id") or f"toolu_{idx}" + name = block.get("name", "") + partial_json = block.get("partial_json", "") + if partial_json: + try: + args = json.loads(partial_json) + except Exception: + args = {"raw": partial_json} + else: + args = block.get("input") or {} + if not isinstance(args, dict): + args = {"input": args} + completed_tool_calls.append( + ToolCall(name=name, args=args, call_id=call_id) + ) + + if completed_tool_calls: + yield StreamChunk(tool_calls=completed_tool_calls) + + # Record assistant response in conversation history + asst_content = [] + for idx in sorted(content_blocks.keys()): + block = content_blocks[idx] + b_type = block.get("type") + if b_type == "text": + txt = block.get("text", "") + if txt: + asst_content.append({"type": "text", "text": txt}) + elif b_type == "thinking": + thk = block.get("thinking", "") + thk_block = {"type": "thinking", "thinking": thk} + if "signature" in block: + thk_block["signature"] = block["signature"] + asst_content.append(thk_block) + elif b_type == "tool_use": + call_id = block.get("id") or f"toolu_{idx}" + name = block.get("name", "") + partial_json = block.get("partial_json", "") + if partial_json: + try: + args = json.loads(partial_json) + except Exception: + args = {"raw": partial_json} + else: + args = block.get("input") or {} + asst_content.append( + { + "type": "tool_use", + "id": call_id, + "name": name, + "input": args, + } + ) + + if asst_content: + self.messages.append({"role": "assistant", "content": asst_content}) + elif accumulated_text: + self.messages.append({"role": "assistant", "content": accumulated_text}) + + +class AnthropicBackend(BaseClient): + def __init__( + self, + config: Optional[Dict[str, Any]] = None, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + headers: Optional[Dict[str, str]] = None, + ): + self.config = config or {} + if api_key is None and config: + api_key = load_backend_api_key(config) + self.api_key = api_key + + url = base_url or self.config.get("base_url") or "https://api.anthropic.com/v1" + url = url.rstrip("/") + if not url.endswith("/v1") and "anthropic.com" in url: + url = f"{url}/v1" + self.base_url = url + + self.headers = dict(headers or self.config.get("headers", {})) + self.anthropic_version = self.config.get("anthropic_version") or "2023-06-01" + self.max_tokens = int(self.config.get("max_tokens") or 4096) + + def _make_request( + self, endpoint: str, data: Optional[Dict[str, Any]] = None, stream: bool = False + ): + url = f"{self.base_url}{endpoint}" + req_headers = { + "Content-Type": "application/json", + "anthropic-version": self.anthropic_version, + "User-Agent": "Vimini/AnthropicAPI", + } + if self.api_key: + req_headers["x-api-key"] = self.api_key + req_headers.update(self.headers) + + body_bytes = json.dumps(data).encode("utf-8") if data is not None else None + req = urllib.request.Request( + url, + data=body_bytes, + headers=req_headers, + method="POST" if data is not None else "GET", + ) + + try: + resp = urllib.request.urlopen(req, timeout=120 if stream else 60) + if stream: + return resp + resp_bytes = resp.read() + return json.loads(resp_bytes.decode("utf-8")) + except urllib.error.HTTPError as e: + err_body = e.read().decode("utf-8", errors="replace") + logger.error(f"Anthropic HTTP error {e.code} for {url}: {err_body}") + raise RuntimeError(f"Anthropic HTTP error {e.code}: {err_body}") from e + except Exception as e: + logger.error(f"Anthropic request error for {url}: {e}") + raise RuntimeError(f"Anthropic connection error: {e}") from e + + def _read_sse_stream(self, resp) -> Iterator[Dict[str, Any]]: + buffer = "" + + def parse_event(event_text: str): + data_lines = [] + for raw_event_line in event_text.splitlines(): + event_line = raw_event_line.rstrip("\r") + if not event_line or event_line.startswith(":"): + continue + if event_line.startswith("data:"): + data_lines.append(event_line[5:].lstrip()) + + if not data_lines: + return None + + data_str = "\n".join(data_lines) + try: + return json.loads(data_str) + except json.JSONDecodeError: + return None + + try: + for raw_line in resp: + line = raw_line.decode("utf-8", errors="replace") + buffer += line + normalized = buffer.replace("\r\n", "\n").replace("\r", "\n") + while "\n\n" in normalized: + event_text, normalized = normalized.split("\n\n", 1) + event = parse_event(event_text) + if event is not None: + if event.get("type") == "error": + err = event.get("error", {}) + msg = err.get("message") or str(err) + raise RuntimeError(f"Anthropic streaming error: {msg}") + yield event + if event.get("type") == "message_stop": + return + buffer = normalized + + if buffer.strip(): + event = parse_event(buffer) + if event is not None: + if event.get("type") == "error": + err = event.get("error", {}) + msg = err.get("message") or str(err) + raise RuntimeError(f"Anthropic streaming error: {msg}") + yield event + finally: + try: + resp.close() + except Exception: + pass + + def list_models(self) -> List[Dict[str, Any]]: + data = self._make_request("/models", stream=False) + raw_models = data if isinstance(data, list) else data.get("data", []) + models: List[Dict[str, Any]] = [] + for m in raw_models: + if isinstance(m, str): + m_id = m + disp_name = m + desc = "" + elif isinstance(m, dict): + m_id = m.get("id") or m.get("name") + disp_name = m.get("display_name") or m.get("name") or m_id + desc = m.get("description", "") + else: + continue + + if not m_id: + continue + + models.append( + { + "name": m_id, + "display_name": disp_name, + "description": desc, + "supported_actions": ["generateContent"], + } + ) + return models + + def generate_content( + self, + prompt: Any, + model: str, + system_instruction: Optional[str] = None, + verbose: bool = False, + response_schema: Optional[Any] = None, + config: Optional[Any] = None, + ) -> str: + if isinstance(prompt, list): + prompt_str = "\n\n".join(str(p) for p in prompt if p) + elif isinstance(prompt, str): + prompt_str = prompt + else: + prompt_str = str(prompt) + + payload: Dict[str, Any] = { + "model": model, + "max_tokens": self.max_tokens, + "messages": [{"role": "user", "content": prompt_str}], + "stream": False, + } + if system_instruction: + payload["system"] = system_instruction + + data = self._make_request("/messages", data=payload, stream=False) + content = data.get("content", []) + if isinstance(content, list): + text_parts = [ + b.get("text", "") + for b in content + if isinstance(b, dict) and b.get("type") == "text" + ] + return "".join(text_parts) + elif isinstance(content, str): + return content + return "" + + def generate_content_stream( + self, + prompt: Any, + model: str, + system_instruction: Optional[str] = None, + verbose: bool = False, + response_schema: Optional[Any] = None, + config: Optional[Any] = None, + ) -> Iterator[StreamChunk]: + if isinstance(prompt, list): + prompt_str = "\n\n".join(str(p) for p in prompt if p) + elif isinstance(prompt, str): + prompt_str = prompt + else: + prompt_str = str(prompt) + + payload: Dict[str, Any] = { + "model": model, + "max_tokens": self.max_tokens, + "messages": [{"role": "user", "content": prompt_str}], + "stream": True, + } + if system_instruction: + payload["system"] = system_instruction + + resp = self._make_request("/messages", data=payload, stream=True) + for event_data in self._read_sse_stream(resp): + event_type = event_data.get("type") + if event_type == "content_block_delta": + delta = event_data.get("delta", {}) + delta_type = delta.get("type") + text_chunk = "" + thought_chunk = "" + if delta_type == "text_delta": + text_chunk = delta.get("text", "") or "" + elif delta_type == "thinking_delta": + thought_chunk = delta.get("thinking", "") or "" + if text_chunk or thought_chunk: + yield StreamChunk(text=text_chunk, thought=thought_chunk) + + def create_chat( + self, + model: str, + system_instruction: Optional[str] = None, + tools: Optional[Any] = None, + history: Optional[Any] = None, + verbose: bool = False, + config: Optional[Any] = None, + ) -> BaseChatSession: + anthropic_tools = convert_tools_to_anthropic(tools) + return AnthropicChatSession( + backend=self, + model=model, + system_instruction=system_instruction, + tools=anthropic_tools, + ) diff --git a/python3/vimini/common/config.py b/python3/vimini/common/config.py index 16cc935..cb00495 100644 --- a/python3/vimini/common/config.py +++ b/python3/vimini/common/config.py @@ -9,7 +9,9 @@ DEFAULT_BACKEND = "Gemini" DEFAULT_MODEL = "gemini-3.8-flash" DEFAULT_GEMINI_KEY_FILE = "~/.config/gemini.token" -SUPPORTED_API_TYPES = {"genai", "openai"} +DEFAULT_ANTHROPIC_KEY_FILE = "~/.config/anthropic.token" +DEFAULT_ANTHROPIC_MODEL = "claude-3-7-sonnet-20250219" +SUPPORTED_API_TYPES = {"genai", "openai", "anthropic"} DEFAULT_UI_CONFIG = { "thinking_color": "DarkGray", @@ -37,15 +39,15 @@ def backend_requires_api_key(backend_config): """ Returns whether a backend requires an API key. - Gemini always requires an API key. OpenAI-compatible endpoints default to - requiring one if api_key_file is defined and non-empty, and not requiring one - if omitted (e.g. for local keyless endpoints). Explicit requires_api_key - takes precedence if specified. + Gemini and Anthropic always require an API key by default. OpenAI-compatible + endpoints default to requiring one if api_key_file is defined and non-empty, + and not requiring one if omitted (e.g. for local keyless endpoints). Explicit + requires_api_key takes precedence if specified. """ if not isinstance(backend_config, dict): return True api = str(backend_config.get("api", "genai")).strip() - default = api == "genai" or bool(backend_config.get("api_key_file")) + default = api in ("genai", "anthropic") or bool(backend_config.get("api_key_file")) return _to_bool(backend_config.get("requires_api_key"), default=default) @@ -123,7 +125,7 @@ def validate_global_config(config): f"Backend '{name}' must specify a non-empty 'api_key_file'" ) default_model = backend_info.get("default_model") - if api == "openai" and ( + if api in ("openai", "anthropic") and ( not isinstance(default_model, str) or not default_model.strip() ): raise ValueError( diff --git a/python3/vimini/config.py b/python3/vimini/config.py index 8934451..600de09 100644 --- a/python3/vimini/config.py +++ b/python3/vimini/config.py @@ -107,7 +107,7 @@ def _ordered_backend_fields(backend_cfg): for key in ("api", "api_key_file", "default_model"): if key in backend_cfg or key == "api": fields.append(key) - if backend_cfg.get("api") == "openai" or "base_url" in backend_cfg: + if backend_cfg.get("api") in ("openai", "anthropic") or "base_url" in backend_cfg: fields.append("base_url") for key in backend_cfg.keys(): if key not in fields: @@ -440,7 +440,9 @@ def add_global_backend(): util.display_message(f"Backend '{name}' already exists.", error=True) return api = ( - _to_str(vim.eval("input('API type [genai/openai]: ', 'openai')")).strip() + _to_str( + vim.eval("input('API type [genai/openai/anthropic]: ', 'openai')") + ).strip() or "openai" ) default_model = _to_str(vim.eval("input('Default model: ')")).strip() @@ -450,7 +452,7 @@ def add_global_backend(): backend_cfg["api_key_file"] = api_key_file if default_model: backend_cfg["default_model"] = default_model - if api == "openai": + if api in ("openai", "anthropic"): base_url = _to_str(vim.eval("input('Base URL: ')")).strip() if base_url: backend_cfg["base_url"] = base_url diff --git a/python3/vimini/main.py b/python3/vimini/main.py index 5b4774f..c6ba877 100644 --- a/python3/vimini/main.py +++ b/python3/vimini/main.py @@ -36,14 +36,14 @@ def _get_backend_default_model(backend_name, backend_cfg, default_model): Returns the configured model for a backend. GenAI backends can fall back to the built-in Gemini default for legacy - configs. OpenAI-compatible backends must name an explicit model so we do - not accidentally send Gemini model names to non-Gemini providers. + configs. OpenAI-compatible and Anthropic backends must name an explicit model + so we do not accidentally send Gemini model names to non-Gemini providers. """ configured_model = backend_cfg.get("default_model") if backend_cfg else None if configured_model: return configured_model api = str((backend_cfg or {}).get("api", "genai")).strip() - if api == "openai": + if api in ("openai", "anthropic"): raise ValueError( f"Backend '{backend_name}' must specify a non-empty 'default_model'" ) diff --git a/python3/vimini/models.py b/python3/vimini/models.py index e061e08..faacfdf 100644 --- a/python3/vimini/models.py +++ b/python3/vimini/models.py @@ -135,6 +135,7 @@ def categorize_model(model): or name_lower.startswith("gemini") or name_lower.startswith("gemma") or name_lower.startswith("learnlm") + or name_lower.startswith("claude") ): return SUPPORTED_CATEGORY @@ -175,7 +176,7 @@ def build_models_buffer(models, current_model, backend_name=None, api_type=None) if current_model and current_model.startswith("models/"): current_model = current_model[7:] - if api_type == "openai": + if api_type in ("openai", "anthropic"): cat = "Available Models" buffer_lines.append(f"[{cat}]") line_map[len(buffer_lines)] = None diff --git a/tests/test_anthropic_backend.py b/tests/test_anthropic_backend.py new file mode 100644 index 0000000..952c184 --- /dev/null +++ b/tests/test_anthropic_backend.py @@ -0,0 +1,533 @@ +import io +import json +import os +import sys +import pytest +from unittest.mock import MagicMock, patch +from urllib.error import HTTPError + +sys.path.insert( + 0, os.path.realpath(os.path.join(os.path.dirname(__file__), "..", "python3")) +) + +from google.genai import types +from vimini.common.backends.base import ToolCall, ToolResponse +from vimini.common.backends.anthropic_backend import ( + AnthropicBackend, + AnthropicChatSession, + convert_schema_to_anthropic, + convert_tools_to_anthropic, +) +from vimini.common.backends import get_backend + + +def test_convert_schema_to_anthropic(): + schema = types.Schema( + type=types.Type.OBJECT, + description="A test object", + properties={ + "path": types.Schema(type=types.Type.STRING, description="File path"), + "count": types.Schema(type=types.Type.INTEGER, enum=["1", "2"]), + }, + required=["path"], + ) + + converted = convert_schema_to_anthropic(schema) + assert converted["type"] == "object" + assert converted["description"] == "A test object" + assert "path" in converted["properties"] + assert converted["properties"]["path"]["type"] == "string" + assert converted["properties"]["path"]["description"] == "File path" + assert converted["properties"]["count"]["type"] == "integer" + assert converted["properties"]["count"]["enum"] == ["1", "2"] + assert converted["required"] == ["path"] + + +def test_convert_schema_none_and_dict(): + assert convert_schema_to_anthropic(None) == {"type": "object", "properties": {}} + + dict_schema = {"type": "STRING", "description": "test"} + conv = convert_schema_to_anthropic(dict_schema) + assert conv["type"] == "string" + assert conv["description"] == "test" + + +def test_convert_tools_to_anthropic(): + genai_tool = types.Tool( + function_declarations=[ + types.FunctionDeclaration( + name="read_file", + description="Reads file", + parameters=types.Schema( + type=types.Type.OBJECT, + properties={"filepath": types.Schema(type=types.Type.STRING)}, + required=["filepath"], + ), + ) + ] + ) + + anthropic_tools = convert_tools_to_anthropic([genai_tool]) + assert len(anthropic_tools) == 1 + tool = anthropic_tools[0] + assert tool["name"] == "read_file" + assert tool["description"] == "Reads file" + assert tool["input_schema"]["type"] == "object" + assert "filepath" in tool["input_schema"]["properties"] + + +def test_convert_tools_dict_and_openai_format(): + dict_tool = { + "type": "function", + "function": { + "name": "test_cmd", + "description": "Runs test", + "parameters": {"type": "object", "properties": {}}, + }, + } + tools = convert_tools_to_anthropic([dict_tool]) + assert len(tools) == 1 + assert tools[0]["name"] == "test_cmd" + assert tools[0]["description"] == "Runs test" + assert tools[0]["input_schema"]["type"] == "object" + + anthropic_native_tool = { + "name": "direct_tool", + "description": "Direct", + "input_schema": {"type": "object"}, + } + tools2 = convert_tools_to_anthropic([anthropic_native_tool]) + assert len(tools2) == 1 + assert tools2[0]["name"] == "direct_tool" + assert tools2[0]["input_schema"] == {"type": "object"} + + +def test_anthropic_backend_init(): + backend = AnthropicBackend( + config={ + "base_url": "https://api.anthropic.com", + "headers": {"X-Custom": "custom-val"}, + "max_tokens": 8192, + "anthropic_version": "2023-06-01", + }, + api_key="sk-ant-test", + ) + assert backend.base_url == "https://api.anthropic.com/v1" + assert backend.api_key == "sk-ant-test" + assert backend.headers == {"X-Custom": "custom-val"} + assert backend.max_tokens == 8192 + assert backend.anthropic_version == "2023-06-01" + + +def test_anthropic_backend_make_request_headers(): + backend = AnthropicBackend( + base_url="https://api.anthropic.com/v1", + api_key="secret-ant-key", + headers={"X-Extra": "extra-val"}, + ) + + mock_resp = MagicMock() + mock_resp.read.return_value = b'{"status": "ok"}' + + with patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen: + res = backend._make_request("/messages", data={"model": "claude-3-5-sonnet"}) + assert res == {"status": "ok"} + + req = mock_urlopen.call_args[0][0] + assert req.full_url == "https://api.anthropic.com/v1/messages" + assert req.headers["X-api-key"] == "secret-ant-key" + assert req.headers["Anthropic-version"] == "2023-06-01" + assert req.headers["X-extra"] == "extra-val" + assert req.headers["Content-type"] == "application/json" + + +def test_anthropic_backend_make_request_error(): + backend = AnthropicBackend() + with patch( + "urllib.request.urlopen", + side_effect=HTTPError( + "url", 401, "Unauthorized", {}, io.BytesIO(b"Invalid API Key") + ), + ): + with pytest.raises( + RuntimeError, match="Anthropic HTTP error 401: Invalid API Key" + ): + backend._make_request("/messages") + + +def test_anthropic_backend_list_models(): + backend = AnthropicBackend() + mock_data = { + "data": [ + { + "id": "claude-3-7-sonnet-20250219", + "display_name": "Claude 3.7 Sonnet", + "description": "Anthropic's latest hybrid reasoning model", + }, + { + "id": "claude-3-5-haiku-20241022", + "display_name": "Claude 3.5 Haiku", + }, + ] + } + + with patch.object(backend, "_make_request", return_value=mock_data): + models = backend.list_models() + assert len(models) == 2 + assert models[0]["name"] == "claude-3-7-sonnet-20250219" + assert models[0]["display_name"] == "Claude 3.7 Sonnet" + assert models[0]["description"] == "Anthropic's latest hybrid reasoning model" + assert models[0]["supported_actions"] == ["generateContent"] + + assert models[1]["name"] == "claude-3-5-haiku-20241022" + assert models[1]["display_name"] == "Claude 3.5 Haiku" + + +def test_anthropic_backend_generate_content(): + backend = AnthropicBackend() + mock_resp_data = { + "content": [ + { + "type": "text", + "text": "Hello from Claude!", + } + ] + } + + with patch.object( + backend, "_make_request", return_value=mock_resp_data + ) as mock_req: + text = backend.generate_content( + "Hello", + model="claude-3-5-sonnet-20241022", + system_instruction="Be helpful", + ) + assert text == "Hello from Claude!" + + payload = mock_req.call_args[1]["data"] + assert payload["model"] == "claude-3-5-sonnet-20241022" + assert payload["system"] == "Be helpful" + assert payload["messages"] == [{"role": "user", "content": "Hello"}] + assert payload["stream"] is False + + +def test_anthropic_backend_generate_content_stream(): + backend = AnthropicBackend() + + sse_lines = [ + b'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking"}}\n\n', + b'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Thinking about it..."}}\n\n', + b'data: {"type": "content_block_stop", "index": 0}\n\n', + b'data: {"type": "content_block_start", "index": 1, "content_block": {"type": "text"}}\n\n', + b'data: {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "Hello"}}\n\n', + b'data: {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": " world"}}\n\n', + b'data: {"type": "content_block_stop", "index": 1}\n\n', + b'data: {"type": "message_stop"}\n\n', + ] + mock_resp = io.BytesIO(b"".join(sse_lines)) + + with patch.object(backend, "_make_request", return_value=mock_resp): + chunks = list( + backend.generate_content_stream("Hi", model="claude-3-5-sonnet-20241022") + ) + + assert len(chunks) == 3 + assert chunks[0].thought == "Thinking about it..." + assert chunks[1].text == "Hello" + assert chunks[2].text == " world" + + +def test_anthropic_read_sse_stream_error_event(): + backend = AnthropicBackend() + sse_lines = [ + b'data: {"type": "error", "error": {"type": "overloaded_error", "message": "Server overloaded"}}\n\n' + ] + mock_resp = io.BytesIO(b"".join(sse_lines)) + + with pytest.raises( + RuntimeError, match="Anthropic streaming error: Server overloaded" + ): + list(backend._read_sse_stream(mock_resp)) + + +def test_anthropic_chat_session_streaming_and_tools(): + backend = AnthropicBackend() + session = backend.create_chat( + model="claude-3-5-sonnet-20241022", + system_instruction="System prompt", + tools=[{"name": "read_file", "parameters": {}}], + ) + + assert isinstance(session, AnthropicChatSession) + assert session.has_history() is False + assert session.get_history() == [] + + # Stream turn 1: text delta followed by tool_use block + events_turn1 = [ + {"type": "message_start", "message": {"id": "msg_1", "role": "assistant"}}, + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "Let me read the file."}, + }, + {"type": "content_block_stop", "index": 0}, + { + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "tool_use", + "id": "toolu_01", + "name": "read_file", + "input": {}, + }, + }, + { + "type": "content_block_delta", + "index": 1, + "delta": { + "type": "input_json_delta", + "partial_json": '{"filepath": "te', + }, + }, + { + "type": "content_block_delta", + "index": 1, + "delta": {"type": "input_json_delta", "partial_json": 'st.py"}'}, + }, + {"type": "content_block_stop", "index": 1}, + {"type": "message_delta", "delta": {"stop_reason": "tool_use"}}, + {"type": "message_stop"}, + ] + + sse_lines_1 = [f"data: {json.dumps(ev)}\n\n".encode("utf-8") for ev in events_turn1] + mock_resp1 = io.BytesIO(b"".join(sse_lines_1)) + + with patch.object(backend, "_make_request", return_value=mock_resp1): + chunks = list(session.send_message_stream("Read test.py")) + + assert len(chunks) == 2 + assert chunks[0].text == "Let me read the file." + assert len(chunks[1].tool_calls) == 1 + tc = chunks[1].tool_calls[0] + assert tc.name == "read_file" + assert tc.args == {"filepath": "test.py"} + assert tc.call_id == "toolu_01" + + # Turn 2: tool response + events_turn2 = [ + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "The file content is valid."}, + }, + {"type": "content_block_stop", "index": 0}, + {"type": "message_stop"}, + ] + sse_lines_2 = [f"data: {json.dumps(ev)}\n\n".encode("utf-8") for ev in events_turn2] + mock_resp2 = io.BytesIO(b"".join(sse_lines_2)) + + tool_res = ToolResponse( + name="read_file", response="print('hello')", call_id="toolu_01" + ) + + with patch.object(backend, "_make_request", return_value=mock_resp2) as mock_req2: + chunks2 = list(session.send_message_stream([tool_res])) + assert len(chunks2) == 1 + assert chunks2[0].text == "The file content is valid." + + call_payload = mock_req2.call_args[1]["data"] + messages = call_payload["messages"] + + # Turn 1 user message + assert messages[0]["role"] == "user" + assert messages[0]["content"] == "Read test.py" + + # Turn 1 assistant message with text and tool_use blocks + assert messages[1]["role"] == "assistant" + assert len(messages[1]["content"]) == 2 + assert messages[1]["content"][0]["type"] == "text" + assert messages[1]["content"][1]["type"] == "tool_use" + assert messages[1]["content"][1]["id"] == "toolu_01" + assert messages[1]["content"][1]["input"] == {"filepath": "test.py"} + + # Turn 2 user message containing tool_result block + assert messages[2]["role"] == "user" + assert len(messages[2]["content"]) == 1 + assert messages[2]["content"][0]["type"] == "tool_result" + assert messages[2]["content"][0]["tool_use_id"] == "toolu_01" + assert messages[2]["content"][0]["content"] == "print('hello')" + + +def test_anthropic_chat_session_multiple_tool_calls(): + backend = AnthropicBackend() + session = backend.create_chat( + model="claude-3-5-sonnet-20241022", + tools=[ + {"name": "read_file", "parameters": {}}, + {"name": "git_grep", "parameters": {}}, + ], + ) + + events_turn = [ + { + "type": "content_block_start", + "index": 0, + "content_block": { + "type": "tool_use", + "id": "toolu_a", + "name": "read_file", + "input": {}, + }, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": { + "type": "input_json_delta", + "partial_json": '{"filepath": "a.txt"}', + }, + }, + {"type": "content_block_stop", "index": 0}, + { + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "tool_use", + "id": "toolu_b", + "name": "git_grep", + "input": {}, + }, + }, + { + "type": "content_block_delta", + "index": 1, + "delta": { + "type": "input_json_delta", + "partial_json": '{"pattern": "TODO"}', + }, + }, + {"type": "content_block_stop", "index": 1}, + {"type": "message_stop"}, + ] + + mock_resp = io.BytesIO( + b"".join([f"data: {json.dumps(ev)}\n\n".encode("utf-8") for ev in events_turn]) + ) + + with patch.object(backend, "_make_request", return_value=mock_resp): + chunks = list(session.send_message_stream("Check both")) + + assert len(chunks) == 1 + assert len(chunks[0].tool_calls) == 2 + assert chunks[0].tool_calls[0].name == "read_file" + assert chunks[0].tool_calls[1].name == "git_grep" + + # Multi-tool response in single user message + resp_sse = io.BytesIO( + b'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "All done"}}\n\ndata: {"type": "message_stop"}\n\n' + ) + with patch.object(backend, "_make_request", return_value=resp_sse) as mock_req: + list( + session.send_message_stream( + [ + ToolResponse( + name="read_file", response="content_a", call_id="toolu_a" + ), + ToolResponse( + name="git_grep", response="content_b", call_id="toolu_b" + ), + ] + ) + ) + + messages = mock_req.call_args[1]["data"]["messages"] + user_msg = messages[2] + assert user_msg["role"] == "user" + assert len(user_msg["content"]) == 2 + assert user_msg["content"][0]["tool_use_id"] == "toolu_a" + assert user_msg["content"][1]["tool_use_id"] == "toolu_b" + + +def test_anthropic_chat_session_consecutive_user_prompts_merged(): + backend = AnthropicBackend() + session = backend.create_chat(model="claude-3-5-sonnet") + + resp1 = io.BytesIO(b'data: {"type": "message_stop"}\n\n') + with patch.object(backend, "_make_request", return_value=resp1): + list(session.send_message_stream("first message")) + + resp2 = io.BytesIO(b'data: {"type": "message_stop"}\n\n') + with patch.object(backend, "_make_request", return_value=resp2) as mock_req: + list(session.send_message_stream("second message")) + + messages = mock_req.call_args[1]["data"]["messages"] + # Merged into single user turn + assert len(messages) == 1 + assert messages[0]["content"] == "first message\n\nsecond message" + + +def test_get_backend_factory_resolves_anthropic_backend(): + config = { + "active_backend": "Claude", + "backends": { + "Claude": { + "api": "anthropic", + "api_key_file": "~/.config/anthropic.token", + "default_model": "claude-3-7-sonnet-20250219", + } + }, + } + + with patch( + "vimini.common.backends.load_backend_api_key", return_value="fake-ant-key" + ): + backend = get_backend("Claude", config=config) + assert isinstance(backend, AnthropicBackend) + assert backend.base_url == "https://api.anthropic.com/v1" + assert backend.api_key == "fake-ant-key" + + +def test_anthropic_backend_missing_api_key_raises(): + config = { + "active_backend": "Claude", + "backends": { + "Claude": { + "api": "anthropic", + "api_key_file": "~/.config/missing.token", + "default_model": "claude-3-7-sonnet-20250219", + } + }, + } + + with patch("vimini.common.backends.load_backend_api_key", return_value=None): + with pytest.raises(RuntimeError, match="API key for backend 'Claude'"): + get_backend("Claude", config=config) + + +def test_anthropic_backend_keyless_allowed(): + config = { + "active_backend": "LocalClaude", + "backends": { + "LocalClaude": { + "api": "anthropic", + "requires_api_key": False, + "base_url": "http://localhost:8000/v1", + "default_model": "claude-local", + } + }, + } + + backend = get_backend("LocalClaude", config=config) + assert isinstance(backend, AnthropicBackend) + assert backend.api_key is None + assert backend.base_url == "http://localhost:8000/v1" diff --git a/tests/test_backend_abstraction.py b/tests/test_backend_abstraction.py index b37c12c..c56bed2 100644 --- a/tests/test_backend_abstraction.py +++ b/tests/test_backend_abstraction.py @@ -16,6 +16,7 @@ ) from vimini.common.backends.genai_backend import GenAIBackend, GenAIChatSession from vimini.common.backends.proxy_backend import ProxyBackend +from vimini.common.backends.anthropic_backend import AnthropicBackend from vimini.common.backends import get_backend @@ -118,6 +119,25 @@ def test_get_backend_openai(): assert isinstance(backend, ProxyBackend) +def test_get_backend_anthropic(): + config = { + "active_backend": "Claude", + "backends": { + "Claude": { + "api": "anthropic", + "api_key_file": "~/.config/anthropic.token", + "default_model": "claude-3-7-sonnet-20250219", + } + }, + } + with patch( + "vimini.common.backends.load_backend_api_key", return_value="fake-ant-key" + ): + backend = get_backend("Claude", config=config) + assert isinstance(backend, AnthropicBackend) + assert backend.api_key == "fake-ant-key" + + def test_get_backend_reports_missing_api_key(): config = { "active_backend": "Custom 1", diff --git a/tests/test_global_config.py b/tests/test_global_config.py index 86ced25..d461d86 100644 --- a/tests/test_global_config.py +++ b/tests/test_global_config.py @@ -127,12 +127,38 @@ def test_validate_global_config_errors(): } ) - with pytest.raises(ValueError, match="api 'anthropic' is not supported"): + with pytest.raises(ValueError, match="api 'unsupported' is not supported"): validate_global_config( { - "active_backend": "Anthropic", + "active_backend": "Unsupported", "backends": { - "Anthropic": { + "Unsupported": { + "api": "unsupported", + "api_key_file": "~/.config/anthropic.token", + } + }, + } + ) + + with pytest.raises(ValueError, match="must specify a non-empty 'api_key_file'"): + validate_global_config( + { + "active_backend": "Claude", + "backends": { + "Claude": { + "api": "anthropic", + "default_model": "claude-3-7-sonnet-20250219", + } + }, + } + ) + + with pytest.raises(ValueError, match="must specify a non-empty 'default_model'"): + validate_global_config( + { + "active_backend": "Claude", + "backends": { + "Claude": { "api": "anthropic", "api_key_file": "~/.config/anthropic.token", } @@ -140,6 +166,19 @@ def test_validate_global_config_errors(): } ) + validate_global_config( + { + "active_backend": "Claude", + "backends": { + "Claude": { + "api": "anthropic", + "api_key_file": "~/.config/anthropic.token", + "default_model": "claude-3-7-sonnet-20250219", + } + }, + } + ) + with pytest.raises(ValueError, match="must specify a non-empty 'api_key_file'"): validate_global_config( { @@ -316,6 +355,16 @@ def test_backend_requires_api_key(): assert ( backend_requires_api_key({"api": "genai", "requires_api_key": False}) is False ) + # Anthropic backends require API key by default + assert backend_requires_api_key({"api": "anthropic"}) is True + assert ( + backend_requires_api_key({"api": "anthropic", "api_key_file": "~/.token"}) + is True + ) + assert ( + backend_requires_api_key({"api": "anthropic", "requires_api_key": False}) + is False + ) # OpenAI without api_key_file defaults to False assert backend_requires_api_key({"api": "openai"}) is False