mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
Sameer Kadam
58aad80d56
* fix: guard integration HTTP requests against SSRF * fix(openclaw): complete fallback validation and base URL handling Address the remaining review findings in the OpenClaw integration. - Strengthen fallback base_url validation to require a non-empty string, valid HTTP(S) scheme, netloc, and hostname. - Strip leading and trailing whitespace from base_url before storing it. - Replace the flaky endpoint-construction test that made a real network connection with mocked session assertions. - Add coverage for _get and _post endpoint construction and timeout forwarding. - Add regression tests for whitespace-padded base URLs and the fallback validation path. These changes complete the Qodo review fixes and harden OpenClaw URL handling without changing the intended localhost/private deployment behavior.
288 lines
11 KiB
Python
288 lines
11 KiB
Python
"""
|
|
OpenClaw ↔ Semantica bridge
|
|
============================
|
|
|
|
Two integration paths:
|
|
|
|
1. **MCP (recommended)** — ``OpenClawMCPConfig`` emits the ``mcporter.json``
|
|
snippet that wires Semantica's MCP server into the OpenClaw Gateway.
|
|
All 12 Semantica MCP tools become native OpenClaw agent tools with no
|
|
extra code.
|
|
|
|
2. **REST** — ``OpenClawKGTool`` is a plain Python class that calls the
|
|
Semantica REST API (port 8000) and can be registered as an OpenClaw
|
|
native tool via SOUL.md ``tools:`` entries.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MCP config helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class OpenClawMCPConfig:
|
|
"""
|
|
Generates the ``mcporter.json`` entry needed to connect Semantica's MCP
|
|
server to the OpenClaw Gateway.
|
|
|
|
Parameters
|
|
----------
|
|
server_command:
|
|
Shell command used to launch the Semantica MCP server.
|
|
Defaults to ``"python -m semantica.mcp_server"``.
|
|
transport:
|
|
MCP transport protocol. OpenClaw supports ``"stdio"`` (default)
|
|
and ``"sse"``.
|
|
name:
|
|
Key used in ``mcporter.json``. Defaults to ``"semantica"``.
|
|
|
|
Example
|
|
-------
|
|
>>> cfg = OpenClawMCPConfig()
|
|
>>> print(cfg.to_json())
|
|
# → paste into ~/.openclaw/mcporter.json, then:
|
|
# → openclaw gateway restart
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
server_command: str = "python -m semantica.mcp_server",
|
|
transport: str = "stdio",
|
|
name: str = "semantica",
|
|
) -> None:
|
|
self.server_command = server_command
|
|
self.transport = transport
|
|
self.name = name
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Return the config as a plain dict."""
|
|
parts = self.server_command.split()
|
|
return {
|
|
"mcpServers": {
|
|
self.name: {
|
|
"command": parts[0],
|
|
"args": parts[1:],
|
|
"transport": self.transport,
|
|
}
|
|
}
|
|
}
|
|
|
|
def to_json(self, indent: int = 2) -> str:
|
|
"""Return the config as a JSON string."""
|
|
return json.dumps(self.to_dict(), indent=indent)
|
|
|
|
def __repr__(self) -> str: # pragma: no cover
|
|
return f"OpenClawMCPConfig(name={self.name!r}, transport={self.transport!r})"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REST-based native tool
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class OpenClawKGTool:
|
|
"""
|
|
A Semantica knowledge-graph tool callable from an OpenClaw agent.
|
|
|
|
Wraps the Semantica REST API so that an OpenClaw agent configured with
|
|
this tool (via SOUL.md ``tools:`` entries or programmatic registration)
|
|
can extract entities, record decisions, query the graph, and more —
|
|
without requiring the MCP gateway.
|
|
|
|
Parameters
|
|
----------
|
|
base_url:
|
|
Base URL of the running Semantica REST server.
|
|
Defaults to ``"http://localhost:8000"``.
|
|
timeout:
|
|
Request timeout in seconds. Defaults to ``30``.
|
|
|
|
Notes
|
|
-----
|
|
``requests`` is used for HTTP calls. It is listed as an optional
|
|
dependency under ``semantica[openclaw]``; install it with::
|
|
|
|
pip install semantica[openclaw]
|
|
"""
|
|
|
|
TOOL_NAME = "semantica_kg"
|
|
TOOL_DESCRIPTION = (
|
|
"Semantica knowledge-graph tool. "
|
|
"Supports entity extraction, decision recording, graph querying, "
|
|
"causal chain analysis, reasoning, and multi-format export."
|
|
)
|
|
|
|
def __init__(self, base_url: str = "http://localhost:8000", timeout: int = 30) -> None:
|
|
# Validate base_url at construction time so callers get an immediate,
|
|
# actionable error rather than a cryptic failure on the first request.
|
|
# allow_private_ips=True because the documented default (localhost:8000)
|
|
# is intentionally a local Semantica server; the scheme check and
|
|
# URL-structure check still apply unconditionally.
|
|
try:
|
|
from semantica.ingest.ssrf import validate_url_for_request
|
|
validate_url_for_request(base_url, allow_private_ips=True)
|
|
except ImportError:
|
|
# semantica.ingest not installed in minimal openclaw-only environments;
|
|
# mirror the structural checks that validate_url_for_request performs
|
|
# unconditionally (before allow_private_ips is consulted), so the
|
|
# guarantee in the comment above — "scheme check and URL-structure check
|
|
# still apply unconditionally" — holds in this path too.
|
|
from urllib.parse import urlparse as _urlparse
|
|
if not isinstance(base_url, str) or not base_url.strip():
|
|
raise ValueError("OpenClawKGTool base_url must be a non-empty string.")
|
|
_parsed = _urlparse(base_url.strip())
|
|
_scheme = (_parsed.scheme or "").lower()
|
|
if _scheme not in ("http", "https"):
|
|
raise ValueError(
|
|
f"OpenClawKGTool base_url scheme '{_parsed.scheme}' is not permitted. "
|
|
"Only http and https are allowed."
|
|
)
|
|
if not _parsed.netloc:
|
|
raise ValueError(
|
|
f"Invalid OpenClawKGTool base_url '{base_url}': "
|
|
"URL must include a netloc (domain or host)."
|
|
)
|
|
if not _parsed.hostname:
|
|
raise ValueError(
|
|
f"Invalid OpenClawKGTool base_url '{base_url}': "
|
|
"URL must include a hostname."
|
|
)
|
|
self.base_url = base_url.strip().rstrip("/")
|
|
self.timeout = timeout
|
|
self._session: Any = None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _get_session(self) -> Any:
|
|
if self._session is None:
|
|
try:
|
|
import requests
|
|
self._session = requests.Session()
|
|
except ImportError as exc:
|
|
raise ImportError(
|
|
"The 'requests' package is required for OpenClawKGTool. "
|
|
"Install it with: pip install semantica[openclaw]"
|
|
) from exc
|
|
return self._session
|
|
|
|
def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
session = self._get_session()
|
|
url = f"{self.base_url}{endpoint}"
|
|
response = session.post(url, json=payload, timeout=self.timeout)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def _get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
session = self._get_session()
|
|
url = f"{self.base_url}{endpoint}"
|
|
response = session.get(url, params=params or {}, timeout=self.timeout)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Extraction
|
|
# ------------------------------------------------------------------
|
|
|
|
def extract(self, text: str) -> Dict[str, Any]:
|
|
"""Extract entities and relations from *text*."""
|
|
return self._post("/extract", {"text": text})
|
|
|
|
def extract_entities(self, text: str) -> List[Dict[str, Any]]:
|
|
"""Return only the entity list from *text*."""
|
|
result = self.extract(text)
|
|
return result.get("entities", [])
|
|
|
|
def extract_relations(self, text: str) -> List[Dict[str, Any]]:
|
|
"""Return only the relation list from *text*."""
|
|
result = self.extract(text)
|
|
return result.get("relations", [])
|
|
|
|
# ------------------------------------------------------------------
|
|
# Graph mutation
|
|
# ------------------------------------------------------------------
|
|
|
|
def add_entity(self, label: str, entity_type: str = "Entity", **properties: Any) -> Dict[str, Any]:
|
|
"""Add a node to the knowledge graph."""
|
|
return self._post("/entities", {"label": label, "type": entity_type, **properties})
|
|
|
|
def add_relationship(
|
|
self,
|
|
source: str,
|
|
target: str,
|
|
relation_type: str,
|
|
**properties: Any,
|
|
) -> Dict[str, Any]:
|
|
"""Add an edge between *source* and *target*."""
|
|
return self._post(
|
|
"/relationships",
|
|
{"source": source, "target": target, "type": relation_type, **properties},
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Decisions
|
|
# ------------------------------------------------------------------
|
|
|
|
def record_decision(
|
|
self,
|
|
decision_text: str,
|
|
context: Optional[str] = None,
|
|
**metadata: Any,
|
|
) -> Dict[str, Any]:
|
|
"""Record a decision in the graph."""
|
|
payload: Dict[str, Any] = {"decision": decision_text}
|
|
if context:
|
|
payload["context"] = context
|
|
payload.update(metadata)
|
|
return self._post("/decisions", payload)
|
|
|
|
def query_decisions(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""Search recorded decisions."""
|
|
result = self._get("/decisions/search", {"q": query, "limit": limit})
|
|
return result.get("decisions", [])
|
|
|
|
def find_precedents(self, decision_text: str, top_k: int = 5) -> List[Dict[str, Any]]:
|
|
"""Find past decisions similar to *decision_text*."""
|
|
result = self._post("/decisions/precedents", {"decision": decision_text, "top_k": top_k})
|
|
return result.get("precedents", [])
|
|
|
|
# ------------------------------------------------------------------
|
|
# Analytics & reasoning
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_causal_chain(self, node_id: str, depth: int = 3) -> Dict[str, Any]:
|
|
"""Retrieve the causal chain rooted at *node_id*."""
|
|
return self._get("/causal-chain", {"node_id": node_id, "depth": depth})
|
|
|
|
def run_reasoning(self, rules: List[str], facts: List[str]) -> Dict[str, Any]:
|
|
"""Run the Semantica forward-chaining reasoner."""
|
|
return self._post("/reason", {"rules": rules, "facts": facts})
|
|
|
|
def get_graph_analytics(self) -> Dict[str, Any]:
|
|
"""Return graph-level analytics (centrality, communities, etc.)."""
|
|
return self._get("/analytics")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Export
|
|
# ------------------------------------------------------------------
|
|
|
|
def export_graph(self, fmt: str = "json") -> str:
|
|
"""Export the graph in *fmt* (``json``, ``ttl``, ``graphml``, …)."""
|
|
result = self._get("/export", {"format": fmt})
|
|
return result.get("data", "")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Summary
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_graph_summary(self) -> Dict[str, Any]:
|
|
"""Return a high-level summary of the current graph."""
|
|
return self._get("/graph/summary")
|
|
|
|
def __repr__(self) -> str: # pragma: no cover
|
|
return f"OpenClawKGTool(base_url={self.base_url!r})"
|