Coverage for integrations / agent_engine / tool_allowlist.py: 89.3%
84 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Tool Allowlist by Model Tier
4Restricts which tools are available to each model tier.
5FAST models get read-only tools, BALANCED gets read-write,
6EXPERT gets unrestricted access.
8Unknown models fail closed (no tools allowed).
10Usage:
11 from integrations.agent_engine.tool_allowlist import filter_tools_for_model, check_tool_allowed
13 tools = filter_tools_for_model('groq-llama', all_tools)
14 allowed, reason = check_tool_allowed('groq-llama', 'write_file')
15"""
17import logging
18from typing import List, Optional, Tuple
20logger = logging.getLogger('hevolve_social')
22# Lazy import to avoid circular dependencies
23_ModelTier = None
26def _get_model_tier():
27 global _ModelTier
28 if _ModelTier is None:
29 from integrations.agent_engine.model_registry import ModelTier
30 _ModelTier = ModelTier
31 return _ModelTier
34# Read-only tools safe for fast/cheap models
35_FAST_TOOLS = frozenset({
36 'web_search', 'read_file', 'list_files', 'memory_search',
37 'embeddings_query', 'get_time', 'calculator', 'status_check',
38 'get_weather', 'search_docs', 'get_agent_info',
39})
41# Read-write tools for balanced models
42_BALANCED_TOOLS = _FAST_TOOLS | frozenset({
43 'write_file', 'send_message', 'create_task', 'update_task',
44 'post_content', 'schedule_job', 'send_notification',
45})
47# Expert = None (unrestricted)
48_TIER_TOOLS = None # Populated lazily
51def _get_tier_tools() -> dict:
52 """Lazy-init tier→tool mapping (avoids import-time ModelTier resolution)."""
53 global _TIER_TOOLS
54 if _TIER_TOOLS is not None:
55 return _TIER_TOOLS
57 ModelTier = _get_model_tier()
58 _TIER_TOOLS = {
59 ModelTier.FAST: _FAST_TOOLS,
60 ModelTier.BALANCED: _BALANCED_TOOLS,
61 ModelTier.EXPERT: None, # None = unrestricted
62 }
63 return _TIER_TOOLS
66def _resolve_tier(model_id: str):
67 """Resolve model ID to its tier. Returns None if unknown."""
68 try:
69 from integrations.agent_engine.model_registry import model_registry
70 info = model_registry.get(model_id)
71 if info:
72 return info.get('tier') or info.get('model_tier')
73 except Exception:
74 pass
75 return None
78def filter_tools_for_model(model_id: str, tools: List[dict]) -> List[dict]:
79 """
80 Filter a tool list by model tier.
82 Args:
83 model_id: Model identifier (e.g. 'groq-llama', 'gpt-4.1')
84 tools: List of tool dicts (must have 'name' key)
86 Returns:
87 Filtered list. Expert tier returns all tools.
88 Unknown model returns empty list (fail-closed).
89 """
90 tier = _resolve_tier(model_id)
91 if tier is None:
92 logger.warning(f"Tool allowlist: unknown model '{model_id}', fail-closed (no tools)")
93 return []
95 tier_tools = _get_tier_tools()
96 allowed_set = tier_tools.get(tier)
97 if allowed_set is None:
98 return tools # Expert = unrestricted
100 filtered = [t for t in tools if t.get('name') in allowed_set]
101 if len(filtered) < len(tools):
102 blocked = [t.get('name') for t in tools if t.get('name') not in allowed_set]
103 logger.info(f"Tool allowlist: {model_id} (tier={tier.value}) blocked tools: {blocked}")
104 return filtered
107def check_tool_allowed(model_id: str, tool_name: str) -> Tuple[bool, str]:
108 """
109 Gate function: check if a specific tool is allowed for a model.
111 Returns:
112 (allowed, reason)
113 """
114 tier = _resolve_tier(model_id)
115 if tier is None:
116 return False, f"Unknown model '{model_id}' — fail-closed"
118 tier_tools = _get_tier_tools()
119 allowed_set = tier_tools.get(tier)
120 if allowed_set is None:
121 return True, f"Model tier {tier.value} has unrestricted access"
123 if tool_name in allowed_set:
124 return True, f"Tool '{tool_name}' allowed for tier {tier.value}"
126 return False, f"Tool '{tool_name}' not allowed for tier {tier.value}"
129# ─── Capability summary for prompt injection ───────────────────────────
130# Each static tool name → ≤3-word phrase the draft prompt can show the
131# user-facing model. Drift guard:
132# tests/unit/test_draft_first_dispatch.py asserts every name in
133# _FAST_TOOLS|_BALANCED_TOOLS has an entry here, so a new tool added
134# without a description fails CI before shipping.
135_TOOL_DESCRIPTIONS: dict = {
136 # _FAST_TOOLS (read-only)
137 'web_search': 'web search',
138 'read_file': 'read files',
139 'list_files': 'browse files',
140 'memory_search': 'recall memory',
141 'embeddings_query': 'semantic search',
142 'get_time': 'current time',
143 'calculator': 'math',
144 'status_check': 'system status',
145 'get_weather': 'weather',
146 'search_docs': 'search docs',
147 'get_agent_info': 'agent info',
148 # _BALANCED_TOOLS adds (read-write)
149 'write_file': 'write files',
150 'send_message': 'send messages',
151 'create_task': 'create tasks',
152 'update_task': 'update tasks',
153 'post_content': 'post content',
154 'schedule_job': 'schedule jobs',
155 'send_notification': 'notifications',
156}
159def get_capability_summary() -> str:
160 """Comma-joined, ≤3-word capability list for the draft prompt.
162 Combines:
163 - Static tools (above), name → ≤3-word phrase
164 - ModelCatalog entries, rolled up by type (tts/stt/vlm/video/audio)
165 so 12 TTS voices show as one phrase, not 12 phrases
166 - MCP servers, by server name (auto-discovered via mcp_registry)
167 - Active channel adapters (auto-discovered via channels.admin.api)
168 - Expert-agent registry, rolled up by category
170 Every dynamic source is wrapped in try/except so a missing or not-
171 yet-loaded subsystem silently drops its slice rather than blocking
172 the prompt. Result is intended to be ≤100 tokens at typical install.
174 Single source of truth for "what can this assistant do" surfaced to
175 the draft model — when a new MCP server / channel / video model is
176 registered at runtime, it appears in the next call without code
177 changes.
178 """
179 parts: list = []
181 # Static tools (sorted for stable output)
182 for name in sorted(_FAST_TOOLS | _BALANCED_TOOLS):
183 parts.append(_TOOL_DESCRIPTIONS.get(name, name))
185 # ModelCatalog — roll up by type, not per-entry
186 try:
187 from integrations.service_tools.model_catalog import get_catalog
188 cat = get_catalog()
189 type_counts: dict = {}
190 for entry in cat.list_all():
191 t = getattr(entry.model_type, 'value', str(entry.model_type))
192 type_counts[t] = type_counts.get(t, 0) + 1
193 _MODEL_TYPE_PHRASES = {
194 'tts': lambda n: f'TTS ({n} voices)',
195 'stt': lambda n: f'STT ({n} models)',
196 'vlm': lambda n: f'vision ({n} VLMs)',
197 'video_gen': lambda _: 'video generation',
198 'audio_gen': lambda _: 'audio generation',
199 'llm': lambda n: f'LLMs ({n})',
200 }
201 for t, n in type_counts.items():
202 phrase_fn = _MODEL_TYPE_PHRASES.get(t)
203 if phrase_fn:
204 parts.append(phrase_fn(n))
205 except Exception:
206 pass
208 # MCP servers (server names; tools-per-server would blow the budget)
209 try:
210 from integrations.mcp.mcp_integration import mcp_registry
211 for server_name in mcp_registry.servers:
212 parts.append(server_name)
213 except Exception:
214 pass
216 # Channel adapters
217 try:
218 from integrations.channels.admin.api import api as channel_api
219 for ch in channel_api._channels:
220 parts.append(ch)
221 except Exception:
222 pass
224 # Expert agents — category roll-up only (96 individual is too many)
225 try:
226 from integrations.expert_agents.registry import (
227 ExpertAgentRegistry, AgentCategory,
228 )
229 registry = ExpertAgentRegistry()
230 live_cats = [
231 cat.value for cat in AgentCategory
232 if registry.get_agents_by_category(cat)
233 ]
234 if live_cats:
235 shown = ', '.join(live_cats[:5])
236 ellipsis = '…' if len(live_cats) > 5 else ''
237 parts.append(f'domain experts ({shown}{ellipsis})')
238 except Exception:
239 pass
241 return ', '.join(parts)