| 1 | from __future__ import annotations |
| 2 | |
| 3 | import hashlib |
| 4 | import json |
| 5 | import os |
| 6 | import re |
| 7 | from typing import Any |
| 8 | |
| 9 | from helpers import files, subagents, tool_policy |
| 10 | |
| 11 | |
| 12 | FUNCTION_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$") |
| 13 | TOOL_NAME_EXAMPLE_PATTERN = re.compile( |
| 14 | r"""["']tool_name["']\s*:\s*["']([A-Za-z0-9_-]{1,64})["']""" |
| 15 | ) |
| 16 | TOOL_HEADING_PATTERN = re.compile(r"^\s{0,3}#{1,6}\s+(.+?)\s*$", re.MULTILINE) |
| 17 | TOOL_DECLARATION_PATTERN = re.compile( |
| 18 | r"^\s*-\s+`([A-Za-z0-9_-]{1,64})`:\s+(args?\b.*)$", |
| 19 | re.IGNORECASE | re.MULTILINE, |
| 20 | ) |
| 21 | SIMPLE_ARGS_PATTERN = re.compile( |
| 22 | r"^\s*args?:\s*`([A-Za-z_][A-Za-z0-9_-]*)`\s*$", |
| 23 | re.IGNORECASE | re.MULTILINE, |
| 24 | ) |
| 25 | TOOL_PROMPT_PREFIX = "agent.system.tool." |
| 26 | TOOL_PROMPT_SUFFIX = ".md" |
| 27 | MAX_TOOL_DESCRIPTION_CHARS = 1024 |
| 28 | TOOL_PROMPT_KWARGS_KEY = "_tool_prompt_kwargs" |
| 29 | |
| 30 | |
| 31 | def build_responses_function_tools(agent: Any) -> tuple[list[dict[str, Any]], dict[str, str]]: |
| 32 | """Build permissive Responses function tools from A0 tool prompts and MCP schemas.""" |
| 33 | |
| 34 | tools: list[dict[str, Any]] = [] |
| 35 | name_map: dict[str, str] = {} |
| 36 | |
| 37 | for tool_name, prompt in _local_tool_prompts(agent): |
| 38 | if not tool_policy.resolve_tool(agent, tool_name).allowed: |
| 39 | continue |
| 40 | native_name = _native_tool_name(tool_name) |
| 41 | name_map[native_name] = tool_name |
| 42 | tools.append( |
| 43 | { |
| 44 | "type": "function", |
| 45 | "name": native_name, |
| 46 | "description": _truncate( |
| 47 | tool_policy.tool_prompt_description( |
| 48 | prompt, |
| 49 | tool_name, |
| 50 | fallback=tool_name, |
| 51 | ) |
| 52 | ), |
| 53 | "parameters": _schema_from_prompt(prompt), |
| 54 | } |
| 55 | ) |
| 56 | |
| 57 | for tool_name, tool in _mcp_tools(agent): |
| 58 | if not tool_policy.resolve_tool( |
| 59 | agent, |
| 60 | tool_name, |
| 61 | canonical_id=tool_policy.canonical_mcp_id(tool_name), |
| 62 | ).allowed: |
| 63 | continue |
| 64 | native_name = _native_tool_name(tool_name) |
| 65 | name_map[native_name] = tool_name |
| 66 | tools.append( |
| 67 | { |
| 68 | "type": "function", |
| 69 | "name": native_name, |
| 70 | "description": _truncate(str(tool.get("description") or tool_name)), |
| 71 | "parameters": _schema_from_any(tool.get("input_schema")), |
| 72 | } |
| 73 | ) |
| 74 | |
| 75 | return _dedupe_tools(tools), name_map |
| 76 | |
| 77 | |
| 78 | def original_tool_name(native_name: str, name_map: dict[str, str] | None) -> str: |
| 79 | if not name_map: |
| 80 | return native_name |
| 81 | return name_map.get(native_name, native_name) |
| 82 | |
| 83 | |
| 84 | def _local_tool_prompts(agent: Any) -> list[tuple[str, str]]: |
| 85 | prompt_dirs = subagents.get_paths(agent, "prompts") |
| 86 | tool_files = files.get_unique_filenames_in_dirs( |
| 87 | prompt_dirs, f"{TOOL_PROMPT_PREFIX}*{TOOL_PROMPT_SUFFIX}" |
| 88 | ) |
| 89 | get_data = getattr(agent, "get_data", None) |
| 90 | tool_kwargs = get_data(TOOL_PROMPT_KWARGS_KEY) if callable(get_data) else {} |
| 91 | tool_kwargs = tool_kwargs if isinstance(tool_kwargs, dict) else {} |
| 92 | result: list[tuple[str, str]] = [] |
| 93 | for tool_file in tool_files: |
| 94 | basename = os.path.basename(tool_file) |
| 95 | fallback_name = _tool_name_from_prompt_basename(basename) |
| 96 | if not fallback_name: |
| 97 | continue |
| 98 | try: |
| 99 | prompt = agent.read_prompt(basename, **tool_kwargs.get(basename, {})) |
| 100 | except Exception: |
| 101 | try: |
| 102 | prompt = files.read_file(tool_file) |
| 103 | except Exception: |
| 104 | prompt = "" |
| 105 | for tool_name in _tool_names_from_prompt(prompt, fallback=fallback_name): |
| 106 | if _include_local_tool_prompt(agent, tool_name): |
| 107 | result.append((tool_name, prompt)) |
| 108 | |
| 109 | vision_prompt = _vision_tool_prompt(agent) |
| 110 | if vision_prompt: |
| 111 | result.append(("vision_load", vision_prompt)) |
| 112 | return result |
| 113 | |
| 114 | |
| 115 | def _vision_tool_prompt(agent: Any) -> str: |
| 116 | try: |
| 117 | from plugins._model_config.helpers.model_config import ( |
| 118 | get_chat_model_config, |
| 119 | get_vision_model_config, |
| 120 | ) |
| 121 | |
| 122 | if not ( |
| 123 | get_vision_model_config(agent) |
| 124 | or get_chat_model_config(agent).get("vision", False) |
| 125 | ): |
| 126 | return "" |
| 127 | return agent.read_prompt("agent.system.tools_vision.md") |
| 128 | except Exception: |
| 129 | return "" |
| 130 | |
| 131 | |
| 132 | def _include_local_tool_prompt(agent: Any, tool_name: str) -> bool: |
| 133 | try: |
| 134 | from plugins._a0_connector.helpers.remote_tool_prompts import ( |
| 135 | should_include_remote_tool_prompt, |
| 136 | ) |
| 137 | except Exception: |
| 138 | return True |
| 139 | |
| 140 | return should_include_remote_tool_prompt(agent, tool_name) |
| 141 | |
| 142 | |
| 143 | def _mcp_tools(agent: Any) -> list[tuple[str, dict[str, Any]]]: |
| 144 | try: |
| 145 | import helpers.mcp_handler as mcp_helper |
| 146 | |
| 147 | raw_tools = mcp_helper.MCPConfig.get_for_agent(agent).get_tools() |
| 148 | except Exception: |
| 149 | return [] |
| 150 | |
| 151 | result: list[tuple[str, dict[str, Any]]] = [] |
| 152 | for entry in raw_tools or []: |
| 153 | if not isinstance(entry, dict): |
| 154 | continue |
| 155 | for tool_name, tool in entry.items(): |
| 156 | if isinstance(tool, dict): |
| 157 | result.append((str(tool_name), tool)) |
| 158 | return result |
| 159 | |
| 160 | |
| 161 | def _tool_name_from_prompt_basename(basename: str) -> str: |
| 162 | if not basename.startswith(TOOL_PROMPT_PREFIX) or not basename.endswith( |
| 163 | TOOL_PROMPT_SUFFIX |
| 164 | ): |
| 165 | return "" |
| 166 | name = basename[len(TOOL_PROMPT_PREFIX) : -len(TOOL_PROMPT_SUFFIX)] |
| 167 | if not name or name in {"tools", "tools_vision"}: |
| 168 | return "" |
| 169 | return name |
| 170 | |
| 171 | |
| 172 | def _tool_name_from_prompt(prompt: str, *, fallback: str) -> str: |
| 173 | for match in TOOL_NAME_EXAMPLE_PATTERN.finditer(prompt or ""): |
| 174 | name = match.group(1).strip() |
| 175 | if FUNCTION_NAME_PATTERN.fullmatch(name): |
| 176 | return name |
| 177 | |
| 178 | for match in TOOL_HEADING_PATTERN.finditer(prompt or ""): |
| 179 | name = _tool_name_from_heading(match.group(1)) |
| 180 | if name: |
| 181 | return name |
| 182 | |
| 183 | return fallback |
| 184 | |
| 185 | |
| 186 | def _tool_names_from_prompt(prompt: str, *, fallback: str) -> list[str]: |
| 187 | declarations = [ |
| 188 | match.group(1) for match in TOOL_DECLARATION_PATTERN.finditer(prompt or "") |
| 189 | ] |
| 190 | if declarations: |
| 191 | return list(dict.fromkeys(declarations)) |
| 192 | return [_tool_name_from_prompt(prompt, fallback=fallback)] |
| 193 | |
| 194 | |
| 195 | def _tool_name_from_heading(heading: str) -> str: |
| 196 | token = (heading or "").strip().split(None, 1)[0] if heading else "" |
| 197 | name = token.strip("`'\" :") |
| 198 | if FUNCTION_NAME_PATTERN.fullmatch(name): |
| 199 | return name |
| 200 | return "" |
| 201 | |
| 202 | |
| 203 | def _native_tool_name(tool_name: str) -> str: |
| 204 | if FUNCTION_NAME_PATTERN.fullmatch(tool_name): |
| 205 | return tool_name |
| 206 | slug = re.sub(r"[^A-Za-z0-9_-]+", "_", tool_name).strip("_") |
| 207 | digest = hashlib.sha1(tool_name.encode("utf-8")).hexdigest()[:8] |
| 208 | native = f"{slug[:52]}_{digest}" if slug else f"a0_tool_{digest}" |
| 209 | return native[:64] |
| 210 | |
| 211 | |
| 212 | def _schema_from_prompt(prompt: str) -> dict[str, Any]: |
| 213 | schema = _schema_from_embedded_json(prompt) |
| 214 | if schema: |
| 215 | return schema |
| 216 | match = SIMPLE_ARGS_PATTERN.search(prompt or "") |
| 217 | if match: |
| 218 | return { |
| 219 | "type": "object", |
| 220 | "properties": {match.group(1): {"type": "string"}}, |
| 221 | "additionalProperties": True, |
| 222 | } |
| 223 | return _permissive_schema() |
| 224 | |
| 225 | |
| 226 | def _schema_from_embedded_json(prompt: str) -> dict[str, Any]: |
| 227 | marker = "Input schema for tool_args:" |
| 228 | index = (prompt or "").find(marker) |
| 229 | if index == -1: |
| 230 | return {} |
| 231 | tail = prompt[index + len(marker) :].strip() |
| 232 | candidate = _balanced_json_object(tail) |
| 233 | if not candidate: |
| 234 | return {} |
| 235 | try: |
| 236 | return _schema_from_any(json.loads(candidate)) |
| 237 | except Exception: |
| 238 | return {} |
| 239 | |
| 240 | |
| 241 | def _schema_from_any(schema: Any) -> dict[str, Any]: |
| 242 | if isinstance(schema, dict): |
| 243 | normalized = dict(schema) |
| 244 | normalized.setdefault("type", "object") |
| 245 | if normalized.get("type") == "object" and not isinstance( |
| 246 | normalized.get("properties"), dict |
| 247 | ): |
| 248 | normalized["properties"] = {} |
| 249 | normalized.setdefault("additionalProperties", True) |
| 250 | return normalized |
| 251 | return _permissive_schema() |
| 252 | |
| 253 | |
| 254 | def _permissive_schema() -> dict[str, Any]: |
| 255 | return {"type": "object", "properties": {}, "additionalProperties": True} |
| 256 | |
| 257 | |
| 258 | def _balanced_json_object(text: str) -> str: |
| 259 | start = text.find("{") |
| 260 | if start == -1: |
| 261 | return "" |
| 262 | depth = 0 |
| 263 | in_string = False |
| 264 | escape = False |
| 265 | for index, char in enumerate(text[start:], start=start): |
| 266 | if in_string: |
| 267 | if escape: |
| 268 | escape = False |
| 269 | elif char == "\\": |
| 270 | escape = True |
| 271 | elif char == '"': |
| 272 | in_string = False |
| 273 | continue |
| 274 | if char == '"': |
| 275 | in_string = True |
| 276 | elif char == "{": |
| 277 | depth += 1 |
| 278 | elif char == "}": |
| 279 | depth -= 1 |
| 280 | if depth == 0: |
| 281 | return text[start : index + 1] |
| 282 | return "" |
| 283 | |
| 284 | |
| 285 | def _dedupe_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| 286 | seen: set[str] = set() |
| 287 | result: list[dict[str, Any]] = [] |
| 288 | for tool in tools: |
| 289 | name = str(tool.get("name") or "") |
| 290 | if not name or name in seen: |
| 291 | continue |
| 292 | seen.add(name) |
| 293 | result.append(tool) |
| 294 | return result |
| 295 | |
| 296 | |
| 297 | def _truncate(text: str) -> str: |
| 298 | if len(text) <= MAX_TOOL_DESCRIPTION_CHARS: |
| 299 | return text |
| 300 | return text[: MAX_TOOL_DESCRIPTION_CHARS - 3].rstrip() + "..." |