|
| 1 | +"""Describe tool arguments by field name without reading their values.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +from typing import Any, Dict, List, Optional |
| 6 | + |
| 7 | +from .constants import PostHogMCPAnalyticsProperty |
| 8 | +from .types import InputAliasMap, JsonRecord, ToolInputOptions |
| 9 | + |
| 10 | +_MAX_INPUT_KEYS = 20 |
| 11 | +_MAX_KEY_LENGTH = 64 |
| 12 | +_ANALYTICS_KEYS = {"context", "llm_model", "conversation_id"} |
| 13 | + |
| 14 | + |
| 15 | +def _declared_properties(schema: Any) -> Dict[str, Any]: |
| 16 | + if not isinstance(schema, dict): |
| 17 | + return {} |
| 18 | + properties = schema.get("properties") |
| 19 | + return properties if isinstance(properties, dict) else {} |
| 20 | + |
| 21 | + |
| 22 | +def _alias_names(aliases: Optional[InputAliasMap]) -> List[str]: |
| 23 | + if not isinstance(aliases, dict): |
| 24 | + return [] |
| 25 | + return [ |
| 26 | + name |
| 27 | + for names in aliases.values() |
| 28 | + if isinstance(names, (list, tuple)) |
| 29 | + for name in names |
| 30 | + if isinstance(name, str) |
| 31 | + ] |
| 32 | + |
| 33 | + |
| 34 | +def _aliases_used( |
| 35 | + aliases: Optional[InputAliasMap], input_value: Dict[str, Any] |
| 36 | +) -> List[str]: |
| 37 | + if not isinstance(aliases, dict): |
| 38 | + return [] |
| 39 | + used = [] |
| 40 | + for canonical, names in aliases.items(): |
| 41 | + if not isinstance(canonical, str) or canonical in input_value: |
| 42 | + continue |
| 43 | + names_value: Any = names |
| 44 | + if not isinstance(names_value, (list, tuple)): |
| 45 | + continue |
| 46 | + alias = next( |
| 47 | + ( |
| 48 | + name |
| 49 | + for name in names_value |
| 50 | + if isinstance(name, str) and name in input_value |
| 51 | + ), |
| 52 | + None, |
| 53 | + ) |
| 54 | + if ( |
| 55 | + alias |
| 56 | + and len(alias) <= _MAX_KEY_LENGTH |
| 57 | + and len(canonical) <= _MAX_KEY_LENGTH |
| 58 | + ): |
| 59 | + used.append(f"{alias}:{canonical}") |
| 60 | + return sorted(used)[:_MAX_INPUT_KEYS] |
| 61 | + |
| 62 | + |
| 63 | +def get_tool_input_properties( |
| 64 | + input_value: Any, |
| 65 | + input_schema: Any = None, |
| 66 | + options: Optional[ToolInputOptions] = None, |
| 67 | +) -> JsonRecord: |
| 68 | + """Describe tool arguments without reading their values. |
| 69 | +
|
| 70 | + The schema and aliases must come from the server. Unknown names become one |
| 71 | + ``[redacted]`` entry because an argument name can contain private data. |
| 72 | + """ |
| 73 | + try: |
| 74 | + if type(input_value) is not dict: |
| 75 | + return {} |
| 76 | + resolved = options or ToolInputOptions() |
| 77 | + properties = _declared_properties(input_schema) |
| 78 | + known = set(properties) | set(_alias_names(resolved.input_aliases)) |
| 79 | + keys = [ |
| 80 | + key |
| 81 | + for key in input_value |
| 82 | + if isinstance(key, str) and (key in known or key not in _ANALYTICS_KEYS) |
| 83 | + ] |
| 84 | + declared: List[str] = [] |
| 85 | + undeclared: List[str] = [] |
| 86 | + has_redacted = False |
| 87 | + for key in keys: |
| 88 | + is_declared = key in known |
| 89 | + record = is_declared |
| 90 | + if resolved.should_record_input_key is not None: |
| 91 | + try: |
| 92 | + record = ( |
| 93 | + resolved.should_record_input_key(key, {"declared": is_declared}) |
| 94 | + is True |
| 95 | + ) |
| 96 | + except Exception: # noqa: BLE001 - analytics callbacks fail closed |
| 97 | + record = False |
| 98 | + if len(key) <= _MAX_KEY_LENGTH and record: |
| 99 | + (declared if is_declared else undeclared).append(key) |
| 100 | + else: |
| 101 | + has_redacted = True |
| 102 | + |
| 103 | + visible = [*sorted(declared), *sorted(undeclared)][:_MAX_INPUT_KEYS] |
| 104 | + if has_redacted and len(visible) < _MAX_INPUT_KEYS: |
| 105 | + visible.append("[redacted]") |
| 106 | + aliases_used = _aliases_used(resolved.input_aliases, input_value) |
| 107 | + result: JsonRecord = { |
| 108 | + PostHogMCPAnalyticsProperty.INPUT_KEYS: visible, |
| 109 | + } |
| 110 | + if aliases_used: |
| 111 | + result[PostHogMCPAnalyticsProperty.INPUT_ALIASES_USED] = aliases_used |
| 112 | + return result |
| 113 | + except Exception: # noqa: BLE001 - analytics must not change tool dispatch |
| 114 | + return {} |
0 commit comments