81 lines
3.2 KiB
Python
81 lines
3.2 KiB
Python
"""Tool proxy between agent and MCP: whitelist, budget, prefix filter, trajectory log."""
|
|
import json
|
|
import re
|
|
import time
|
|
|
|
MODEL_TOOLS = {
|
|
"sap_search_object", "sap_pull_source", "sap_object_structure", "sap_object_members",
|
|
"sap_usage_references", "sap_element_info", "sap_inactive_objects", "sap_short_dumps",
|
|
"sap_sql_query", "sap_create_object", "sap_push_source", "sap_push_element",
|
|
"sap_push_message", "sap_activate", "sap_syntax_check", "sap_check_object",
|
|
"sap_run_unit_test", "sap_atc_run", "sap_pretty_print", "sap_run_class",
|
|
}
|
|
WRITE_TOOLS = {"sap_create_object", "sap_push_source", "sap_push_element",
|
|
"sap_push_message", "sap_activate"}
|
|
RUN_PREFIX = re.compile(r"^Z\d[0-9A-Z]{5,6}_", re.I)
|
|
|
|
|
|
class BudgetExceeded(Exception):
|
|
pass
|
|
|
|
|
|
class ToolProxy:
|
|
def __init__(self, mcp, prefix, budget, log_path):
|
|
self.mcp = mcp
|
|
self.prefix = prefix.upper()
|
|
self.max_calls = budget.get("max_tool_calls", 60)
|
|
self.max_activations = budget.get("max_activations", 15)
|
|
self.calls = 0
|
|
self.activations = 0
|
|
self.fail_streak = 0
|
|
self.max_fail_streak = 0
|
|
self.log = open(log_path, "a")
|
|
|
|
def schemas(self):
|
|
return [t for t in self.mcp.list_tools() if t["name"] in MODEL_TOOLS]
|
|
|
|
def _foreign(self, name):
|
|
n = (name or "").upper()
|
|
return bool(RUN_PREFIX.match(n)) and not n.startswith(self.prefix)
|
|
|
|
def _filter(self, tool, text):
|
|
if tool not in ("sap_search_object", "sap_usage_references"):
|
|
return text
|
|
try:
|
|
data = json.loads(text)
|
|
except ValueError:
|
|
return text
|
|
if isinstance(data, list):
|
|
data = [d for d in data if not self._foreign(d.get("name") if isinstance(d, dict) else "")]
|
|
return json.dumps(data)
|
|
return text
|
|
|
|
def call(self, tool, args):
|
|
entry = {"t": time.time(), "tool": tool, "args": args}
|
|
if tool not in MODEL_TOOLS:
|
|
result = (True, f"Tool {tool} is not available.")
|
|
elif self.calls >= self.max_calls:
|
|
raise BudgetExceeded(f"max_tool_calls={self.max_calls}")
|
|
else:
|
|
self.calls += 1
|
|
name = str(args.get("objectName", "")).upper()
|
|
if tool in WRITE_TOOLS and self._foreign(name):
|
|
result = (True, f"{name} is not available.")
|
|
else:
|
|
if tool in WRITE_TOOLS and (tool == "sap_activate" or args.get("activate", True)):
|
|
self.activations += 1
|
|
err, text = self.mcp.call(tool, args)
|
|
result = (err, self._filter(tool, text))
|
|
if tool in WRITE_TOOLS:
|
|
failed = err or '"success":false' in text.replace(" ", "")
|
|
self.fail_streak = self.fail_streak + 1 if failed else 0
|
|
self.max_fail_streak = max(self.max_fail_streak, self.fail_streak)
|
|
entry["is_error"], entry["result"] = result[0], result[1][:20000]
|
|
self.log.write(json.dumps(entry) + "\n")
|
|
self.log.flush()
|
|
return result
|
|
|
|
def note(self, kind, content):
|
|
self.log.write(json.dumps({"t": time.time(), "kind": kind, "content": content}) + "\n")
|
|
self.log.flush()
|