diff --git a/pentestagent/mcp/mcp_servers.json b/pentestagent/mcp/mcp_servers.json index 1779f37..5137914 100644 --- a/pentestagent/mcp/mcp_servers.json +++ b/pentestagent/mcp/mcp_servers.json @@ -3,11 +3,13 @@ "hexstrike-local": { "command": "python3", "args": [ - "third_party/hexstrike/hexstrike_mcp.py", - "--timeout", - "300" + "-u", + "pentestagent/mcp/stdio_adapter.py" ], - "description": "HexStrike AI (vendored) - local MCP adapter (spawned via stdio)", + "env": { + "STDIO_TARGET": "http://127.0.0.1:8888" + }, + "description": "HexStrike (stdio->HTTP) adapter - forwards MCP tools/call to HexStrike HTTP API", "enabled": true, "start_on_launch": false } diff --git a/pentestagent/mcp/stdio_adapter.py b/pentestagent/mcp/stdio_adapter.py index ae804b2..a79319a 100644 --- a/pentestagent/mcp/stdio_adapter.py +++ b/pentestagent/mcp/stdio_adapter.py @@ -32,13 +32,57 @@ except Exception: TARGET = os.environ.get("STDIO_TARGET", "http://127.0.0.1:8888").rstrip("/") _tools_env = os.environ.get("STDIO_TOOLS") +def _default_tools() -> List[Dict[str, str]]: + return [{"name": "http_api", "description": "Generic HTTP proxy"}] + + +def _discover_tools_from_target(target: str) -> List[Dict[str, str]]: + """Attempt to discover tools from the HTTP API at /api/tools. + + The HexStrike server exposes blueprints under `/api/tools` and many + installations provide an index at `/api/tools` returning a JSON list. + If discovery fails, return the default tool list. + """ + if requests is None: + return _default_tools() + try: + url = target.rstrip("/") + "/api/tools" + r = requests.get(url, timeout=10) + if r.status_code != 200: + return _default_tools() + data = r.json() + # Expecting either a list of tools or an object with `tools` key + tools = [] + if isinstance(data, dict) and "tools" in data and isinstance(data["tools"], list): + src = data["tools"] + elif isinstance(data, list): + src = data + else: + return _default_tools() + + for t in src: + # t may be a string or object with name/description + if isinstance(t, str): + tools.append({"name": t, "description": "Remote tool"}) + elif isinstance(t, dict): + name = t.get("name") or t.get("id") or t.get("tool") + desc = t.get("description") or t.get("desc") or "Remote tool" + if name: + tools.append({"name": name, "description": desc}) + if tools: + return tools + except Exception: + pass + return _default_tools() + + if _tools_env: try: TOOLS: List[Dict[str, str]] = json.loads(_tools_env) except Exception: - TOOLS = [{"name": "http_api", "description": "Generic HTTP proxy"}] + TOOLS = _default_tools() else: - TOOLS = [{"name": "http_api", "description": "Generic HTTP proxy"}] + TOOLS = _discover_tools_from_target(TARGET) def _send(resp: Dict[str, Any]) -> None: