github · mcp-server16c78a447a0a

RemediationAI

The tool 'search_datanexus_tools' conflicts with the reserved generic 'search' tool name in the MCP protocol, creating ambiguity for clients. Rename the tool in mcp-manifest.json from 'search_datanexus_tools' to 'datanexus_search' or another server-specific prefix to avoid shadowing reserved semantics. This ensures clients can unambiguously route requests to the correct implementation. Verify by checking that the MCP client can call the renamed tool without protocol conflicts and that no other tools reference the old name.

RemediationAI

The tool 'security_fetch_package_vulnerabilities' reads NVD_API_KEY from os.environ at module level (config.py line 47) without validating caller identity or enforcing per-request credentials. Modify the function signature to accept an optional api_key parameter, and add a check in the tool handler to require the caller to provide credentials via request context or environment override. This ensures each request is authenticated and auditable to a specific caller. Test by invoking the tool with and without credentials and confirming that unauthenticated requests are rejected.

RemediationAI

The tool 'security_fetch_dependency_graph' returns transitive dependency data from deps.dev without any provenance wrapper, source delimiter, or attribution marker, making it impossible for the LLM to distinguish untrusted external data from authoritative sources. Wrap the returned dependency data in a structured object with a 'source' field set to 'deps.dev', a 'fetched_at' timestamp, and a 'provenance' field documenting the external origin. This allows the LLM to reason about data trustworthiness. Verify by checking that the response includes the provenance wrapper and that the source is clearly labeled.

RemediationAI

The tool 'nonprofit_fetch_nonprofit_by_ein' returns IRS 990 data including mission statements and organizational descriptions from external sources (IRS EO BMF, IRS TEOS) without provenance delimiters or source attribution, preventing the LLM from assessing data reliability. Wrap all returned fields in a response object with 'source' (e.g., 'IRS_EO_BMF'), 'fetched_at' timestamp, and 'attribution' fields that cite the external origin. This enables transparent data lineage. Test by verifying that the response includes source attribution for each data field and that the origin is clearly visible.

RemediationAI

The tool 'security_fetch_package_vulnerabilities' returns CVE data from OSV.dev and NIST NVD verbatim without provenance delimiters or source attribution, making it impossible for the LLM to distinguish between authoritative and untrusted sources. Wrap the vulnerability response in a structured object with 'source' field (e.g., 'NIST_NVD' or 'OSV_DEV'), 'fetched_at' timestamp, and 'severity_source' indicating which source provided the CVSS score. This ensures data provenance is transparent. Verify by checking that each vulnerability record includes source attribution and that the LLM can identify the origin of severity scores.

RemediationAI

The tool 'nonprofit_fetch_charity_uk' returns UK charity data including activities and trustee information from the Charity Commission England and Wales without visible provenance wrapper or source attribution, obscuring data origin and reliability. Wrap the returned charity data in a response object with 'source' set to 'Charity_Commission_England_Wales', 'fetched_at' timestamp, and 'data_fields' documenting which fields came from the external source. This provides transparent attribution. Test by confirming that the response includes a provenance wrapper and that the external source is clearly labeled.

Evidence

"""

datanexus/agents/digest_generator.py — Haiku Trigger 4: weekly digest.

Spec: DataNexus_MCP_Spec_v7_4.docx  Section 13.6 / Phase 6

Batches ALL feedback records for a tool into ONE Haiku call (cost control)

and produces a DigestItem with data quality score, top issues, suggested

rules, and sprint recommendations.

Rules (CLAUDE.md):

  S13-1: This is Haiku Trigger T4. Delegate to haiku_classifier.classify() only.

  S13-4: Never raises. Always returns structured DigestItem.

Special case — emp

RemediationAI

The digest_generator tool calls classify() which invokes the Anthropic SDK via anthropic.messages.create() without an explicit max_tokens parameter, allowing unbounded token consumption and cost overruns. Add a max_tokens parameter to the anthropic.messages.create() call in haiku_classifier.classify() with a reasonable cap (e.g., 500 tokens for Haiku), and pass this parameter from digest_generator.py. This enforces a hard token budget per request. Verify by monitoring API costs and confirming that token usage per request does not exceed the configured limit.

Evidence

"""

datanexus/agents/feedback_classifier.py — Haiku Trigger 2: feedback classification.

Spec: DataNexus_MCP_Spec_v7_4.docx  Section 13.4 / Phase 4

Most important file in Section 13. Closes the loop between user-reported

feedback and automated data quality triage.

Rules (CLAUDE.md):

  S13-1: This is Haiku Trigger T2. Delegate to haiku_classifier.classify() only.

  S13-4: Never raises. All exceptions caught. Always returns structured result.

  S13-5: FeedbackRecord.classification is ONE-WAY onl

RemediationAI

The feedback_classifier tool calls classify() which invokes the Anthropic SDK without an explicit max_tokens parameter, allowing unbounded token consumption. Add a max_tokens parameter (e.g., 500 tokens) to the anthropic.messages.create() call in haiku_classifier.classify() and ensure feedback_classifier.py passes this parameter. This prevents cost overruns and enforces predictable resource usage. Test by running the classifier and confirming that token usage stays within the configured budget.

Evidence

"""

datanexus/agents/anomaly_reviewer.py — Haiku Trigger 1: anomaly review.

Spec: DataNexus_MCP_Spec_v7_4.docx  Section 13.3 / Phase 3

Called when the deterministic validator (validator.py) flags an anomaly

that needs a second-opinion classification before deciding whether to

keep, suppress, or flag the affected record.

Rules (CLAUDE.md):

  S13-1: This is Haiku Trigger T1. Do NOT call Anthropic API directly —

         always delegate to haiku_classifier.classify().

  S13-4: Never raises. All 

RemediationAI

The anomaly_reviewer tool calls classify() which invokes the Anthropic SDK without token cap enforcement, risking unbounded costs. Add a max_tokens parameter to anthropic.messages.create() in haiku_classifier.classify() (e.g., 500 tokens) and ensure anomaly_reviewer.py passes this parameter. This enforces a hard token limit per classification. Verify by checking API usage logs and confirming that token consumption per request does not exceed the cap.

Evidence

"""

datanexus/agents/schema_monitor.py — Haiku Trigger 3: schema change assessment.

Spec: DataNexus_MCP_Spec_v7_4.docx  Section 13.5 / Phase 5

Called when the ingest worker detects a structural change between the

previously-stored schema fingerprint and the newly-fetched payload shape.

Haiku decides whether the change is breaking and what severity to assign.

Rules (CLAUDE.md):

  S13-1: This is Haiku Trigger T3. Delegate to haiku_classifier.classify() only.

  S13-4: Never raises. Always return

RemediationAI

The schema_monitor tool calls classify() which invokes the Anthropic SDK without an explicit max_tokens parameter, allowing unbounded token usage. Add a max_tokens parameter (e.g., 500 tokens) to the anthropic.messages.create() call in haiku_classifier.classify() and ensure schema_monitor.py passes this parameter. This enforces predictable token budgets. Test by running schema monitoring and confirming that token usage per request stays within the configured limit.

Evidence

"""

datanexus/agents/haiku_classifier.py — Shared Haiku API wrapper for all 4 triggers.

Spec: DataNexus_MCP_Spec_v7_4.docx  Section 13.2 / Phase 2

Rules (CLAUDE.md):

  S13-1: Called ONLY from the 4 permitted triggers — never directly from tool handlers.

  S13-2: HAIKU_MODEL imported from feedback/config.py. Never hardcoded here.

  S13-3: Daily cap HAIKU_MAX_CALLS_PER_DAY (100) enforced before every call.

         incr → expire → check, in that exact order.

  S13-4: On any API error: return err

RemediationAI

The haiku_classifier.classify() function calls anthropic.messages.create() without an explicit max_tokens parameter, allowing unbounded token consumption across all four triggers (digest_generator, feedback_classifier, anomaly_reviewer, schema_monitor). Add a max_tokens parameter to the anthropic.messages.create() call in haiku_classifier.py (e.g., max_tokens=500) and accept it as a function parameter with a sensible default. This enforces a hard token cap across all callers. Verify by checking that API responses include token usage and that it does not exceed the configured limit.

Evidence

"""

datanexus/tools/security_stateful.py — Sprint 6 stateful security tools.

Tools:

  fetch_cve_watch     — persistent CVE watchlist (create/check/delete)

  audit_sbom_continuous — continuous SBOM monitoring (register/check/deregister)

Redis key schema (dn: prefix):

  dn:cve_watch:{watch_id}    — Hash (watch data)

  dn:cve_watch_ids           — SET  (watch index)

  dn:sbom_watch:{watch_id}   — Hash (SBOM watch data)

  dn:sbom_watch_ids          — SET  (SBOM watch index)

Scheduler: see datanex

RemediationAI

The security_stateful.py tool performs destructive operations (DELETE from watchlist, DEREGISTER from SBOM monitoring) but emits no audit event, leaving no trail for investigators to answer 'who deleted record X on day Y?'. Add an audit logging call (e.g., log_audit_event(action='delete_cve_watch', watch_id=watch_id, user=caller_id, timestamp=now())) immediately after each destructive operation (fetch_cve_watch delete, audit_sbom_continuous deregister). This creates an immutable audit trail. Test by performing a destructive operation and confirming that an audit event is logged with the action, resource ID, caller, and timestamp.

Evidence

# ── Structured error response shape ──────────────────────────────────────────

# Return this dict on ANY error. Never raise. Never return str(e).

# error_code must come from ErrorCode enum below.

#

# {

RemediationAI

The schema.py error handler returns full exception details via str(e), leaking internal paths, library versions, and query structure useful for reconnaissance. Replace str(exc) with a sanitized error message from the ErrorCode enum (e.g., 'SCHEMA_VALIDATION_FAILED') and log the full traceback server-side only. This prevents information disclosure while preserving debugging capability. Verify by triggering an error and confirming that the response contains only a generic error code, not the full stack trace.

Evidence

"error":           str(exc),

            "haiku_available": False,

        }))

        return {"error": str(exc), "haiku_available": False}

RemediationAI

The haiku_classifier.py error handler returns str(exc) to the caller, leaking internal exception details and library versions. Replace str(exc) with a generic error message (e.g., 'Classification service unavailable') and log the full exception server-side only. This prevents information disclosure. Test by triggering a classification error and confirming that the response contains only a generic message, not the full exception traceback.

Evidence

schedule_seconds=_SCHEDULE,

        )

    async def fetch(self) -> bytes:

        """Fetch and cache EPO OPS OAuth token; probe fallback sources."""

        client_id     = os.environ.get("EPO_CLIENT_ID", "")

        client_secret = os.environ.get("EPO_CLIENT_SECRET", "")

RemediationAI

The t11_worker.py fetch() method makes HTTP requests to EPO OPS without an explicit timeout parameter, allowing a hung upstream to pin threads indefinitely. Add timeout=30 (or appropriate value) to the httpx.AsyncClient() or httpx.get() call. This ensures requests fail fast if the upstream is unresponsive. Verify by testing with a slow/hung upstream and confirming that the request times out within the specified duration.

Evidence

schedule_seconds=_SCHEDULE,

        )

    async def fetch(self) -> bytes:

        """

        Fetch NPPES data for top specialities. Returns a summary JSON blob

        for the base class to hash and store. Per-speciality cache entries

RemediationAI

The t22_worker.py fetch() method makes HTTP requests to NPPES without an explicit timeout, risking thread exhaustion if the upstream hangs. Add timeout=30 to the httpx.AsyncClient() or httpx.get() call. This prevents indefinite blocking. Test by simulating a slow upstream and confirming that the request times out within the configured duration.

Evidence

schedule_seconds=_SCHEDULE,

        )

    async def fetch(self) -> bytes:

        """Pre-seed top keyword categories."""

        seeded = 0

        async with httpx.AsyncClient(

RemediationAI

The t18_worker.py fetch() method creates an httpx.AsyncClient() without an explicit timeout parameter, allowing hung upstreams to block indefinitely. Add timeout=30 to the httpx.AsyncClient() constructor. This enforces a bounded wait time. Verify by testing with a slow upstream and confirming that the request times out within the specified duration.

Evidence

self.ttl_seconds      = ttl_seconds

        self.schedule_seconds = schedule_seconds

    async def fetch(self) -> bytes:

        """

        Override in subclass. Must return raw upstream response bytes.

        Raise any exception on failure — run_forever will catch it.

RemediationAI

The IngestBase.fetch() base class documentation does not enforce timeout requirements for subclass implementations, and subclasses may make network calls without timeouts. Add a timeout parameter to the IngestBase class (e.g., timeout_seconds=30) and document that all subclass fetch() implementations must use this timeout when making HTTP or subprocess calls. This ensures all ingest workers respect timeout boundaries. Test by checking that all subclass implementations pass the timeout parameter to their network calls.

Evidence

schedule_seconds=604800,  # weekly — data updates monthly

        )

    async def fetch(self) -> bytes:

        """Download all 4 regional CSVs and index every EIN into Redis."""

        total_rows = 0

RemediationAI

The t04_worker.py fetch() method downloads regional CSVs without an explicit timeout on the HTTP client, risking indefinite blocking. Add timeout=60 to the httpx.AsyncClient() constructor (or increase if needed for large downloads). This prevents thread exhaustion. Verify by testing with a slow upstream and confirming that the download times out within the configured duration.

Evidence

schedule_seconds=_SCHEDULE,  # 21600 — rate limit awareness

        )

    async def fetch(self) -> bytes:

        """Pre-seed top regulatory keyword categories."""

        api_key = os.environ.get("REGULATIONS_GOV_KEY", "")

        if not api_key:

RemediationAI

The t19_worker.py fetch() method makes HTTP requests to Regulations.gov without an explicit timeout, allowing hung upstreams to pin threads. Add timeout=30 to the httpx.AsyncClient() or httpx.get() call. This enforces a bounded wait time. Test by simulating a slow upstream and confirming that the request times out within the specified duration.

Evidence

Downloads from the official CISA GitHub mirror (cisagov/kev-data) — see KEV_URL constant.

and stores in Redis as:

  datanexus:kev:catalog    — full catalog JSON (TTL 25h)

  datanexus:kev:fetched_at — ISO timestamp of last successful fetch (TTL 25h)

  datanexus:kev:last_refresh_error — error message if refresh fails (TTL 7d)

Invoked two ways:

RemediationAI

The kev_refresh.py module downloads from the CISA GitHub mirror without an explicit timeout on the HTTP client, risking indefinite blocking if the upstream is slow or hung. Add timeout=60 to the httpx.get() or httpx.AsyncClient() call. This prevents thread exhaustion. Verify by testing with a slow upstream and confirming that the download times out within the configured duration.

Evidence

Endpoints:

  GET /ops/dashboard          — full HTML dashboard

  GET /ops/api/metrics        — JSON metrics snapshot (same data, for fetch())

  GET /ops/health             — {"ok": true}

The dashboard auto-refreshes every 15 seconds via JavaScript fetch().

RemediationAI

The dashboard.py endpoints make HTTP requests (e.g., via fetch() in JavaScript or backend calls) without explicit timeouts, risking unresponsive dashboard if upstreams hang. Add timeout=30 to any httpx or requests calls in the backend, and set a reasonable fetch() timeout in the JavaScript (e.g., 10 seconds). This ensures the dashboard remains responsive. Test by simulating slow upstreams and confirming that requests time out and the dashboard remains usable.

Evidence

class IngestBase:

    """

    Base ingest worker. Subclasses override fetch() only.

    Args:

        tool_id:          Tool this worker feeds (e.g. 'T04')

RemediationAI

The IngestBase class does not enforce timeout requirements for network or subprocess calls made by subclasses, allowing hung upstreams to block indefinitely. Add a timeout_seconds parameter to IngestBase.__init__() (e.g., timeout_seconds=30) and document that all subclass fetch() implementations must pass this timeout to httpx.AsyncClient(), httpx.get(), subprocess calls, etc. This ensures all ingest workers respect timeout boundaries. Verify by checking that all subclass implementations use the timeout parameter in their network calls.

Evidence

schedule_seconds=_SCHEDULE,

        )

    async def fetch(self) -> bytes:

        """Fetch IANA RDAP bootstrap and store in Redis."""

        async with httpx.AsyncClient(

            timeout=_TIMEOUT, headers=_HEADERS, follow_redirects=True

RemediationAI

The t07_worker.py fetch() method already includes timeout=_TIMEOUT in the httpx.AsyncClient() call, which is correct. Ensure that _TIMEOUT is defined with a reasonable value (e.g., 30 seconds) and that all other network calls in the file also use this timeout. Verify by checking that _TIMEOUT is set to a non-None value and that all HTTP calls use it.

Evidence

schedule_seconds=3600,

        )

    async def fetch(self) -> bytes:

        """Pre-seed all 50 packages. Returns last raw response bytes."""

        seeded = 0

        last_raw = b""

RemediationAI

The t10_worker.py fetch() method pre-seeds packages without an explicit timeout on HTTP calls, risking indefinite blocking. Add timeout=30 to the httpx.AsyncClient() or httpx.get() call. This prevents thread exhaustion. Test by simulating a slow upstream and confirming that the request times out within the configured duration.

Evidence

schedule_seconds=86400,

        )

    async def fetch(self) -> bytes:

        """Download bulk extract and index all main charities into Redis."""

        log.info(json.dumps({

            "ts": _iso_now(), "event": "uk_bulk_download_start", "tool": self.tool_id,

RemediationAI

The t04_worker.py fetch() method downloads bulk charity data without an explicit timeout on the HTTP client, risking indefinite blocking if the upstream is slow. Add timeout=60 to the httpx.AsyncClient() or httpx.get() call (increase if needed for large downloads). This prevents thread exhaustion. Verify by testing with a slow upstream and confirming that the download times out within the configured duration.

Evidence

#!/usr/bin/env node

/**

 * DataNexus MCP — stdio proxy for Glama and local MCP clients.

 *

 * Bridges stdio ↔ the live remote server at https://datanexusmcp.com/mcp

 * using mcp-remote. No Python, no Redis, no PostgreSQL required.

 *

 * Usage:

 *   npx -y @datanexusmcp/mcp-server

 *

 * Claude Desktop claude_desktop_config.json:

 *   {

 *     "mcpServers": {

 *       "datanexus": {

 *         "command": "npx",

 *         "args": ["-y", "@datanexusmcp/mcp-server"]

 *       }

 *     }

 *   }

 */

c

RemediationAI

The datanexus-mcp.js file is a standard remote MCP transport proxy that forwards JSON-RPC to a remote server, which is a legitimate deployment pattern. No remediation is required; this is not a vulnerability. Confirm that the remote server URL (https://datanexusmcp.com/mcp) is hardcoded and not user-controllable, and that the proxy does not accept steering instructions in the JSON-RPC payload.

Evidence

try:

                r = await get_redis()

                if r:

                    await r.set(cache_key, tier, ex=300)

            except Exception:

                pass

            return tier

        except Exception as exc:

RemediationAI

The main.py error handler swallows exceptions with bare pass, discarding errors with no log or metric, blinding incident response. Replace the except Exception: pass block with log.warning(f'Cache set failed: {exc}') to record the failure. This enables troubleshooting and incident response. Test by triggering a Redis error and confirming that a warning is logged.

Evidence

# remaining allowlist domains — any path counts

        for allowed in _PATCH_ALLOWLIST:

            if host == allowed or host.endswith("." + allowed):

                return True

    except Exception:

        pass

    return False

RemediationAI

The cve_sprint7.py allowlist check swallows exceptions with bare pass, hiding real failures in domain parsing. Replace except Exception: pass with log.warning(f'Allowlist check failed: {exc}') to record the error. This enables debugging of allowlist logic. Test by triggering a parsing error and confirming that a warning is logged.

Evidence

raw = r.hget(key_feedback(tool_id, rid), "data")

                    if raw:

                        try:

                            records.append(json.loads(raw))

                        except Exception:

                            pass

            except Exception:

                pass

        results[tool_id] = records

RemediationAI

The feedback/dashboard/server.py error handler swallows exceptions with bare pass when parsing feedback records, hiding data corruption or Redis errors. Replace except Exception: pass with log.warning(f'Failed to parse feedback record {rid}: {exc}') to record the failure. This enables detection of data quality issues. Test by introducing a malformed record and confirming that a warning is logged.

Evidence

try:

        r = _get_redis()

        if r:

            r.set(LAST_ERROR_KEY, msg[:500], ex=7 * 86400)

    except Exception:

        pass

async def refresh() -> bool:

RemediationAI

The kev_refresh.py error handler swallows exceptions with bare pass when writing to Redis, hiding failures to persist error messages. Replace except Exception: pass with log.warning(f'Failed to log refresh error: {exc}') to record the failure. This enables troubleshooting of the refresh process. Test by simulating a Redis error and confirming that a warning is logged.