github · mcp-server7cc56523fd08

Evidence

"""

datanexus/tools/t11.py — T11 Global Patent Intelligence tool.

Spec: DataNexus_MCP_Spec_v7_4.docx  Section 4, T11 entry

Exactly 4 data functions. Shared infrastructure tools (report_feedback,

report_mcpize_link) are registered ONCE in main.py — NOT here.

Data sources:

  Primary:   EPO OPS (Open Patent Services) — ops.epo.org — OAuth client_credentials

             European Patent Office API; 4 GB/month free tier

  Secondary: USPTO PatentsView — api.patentsview.org — no key required

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _set_redis_client(client: Optional[redis_lib.Redis]) -> None:

    """Inject a Redis client — used in tests (e.g. fakeredis). Pass None to reset."""

    global _redis_client

    _redis_client = client

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _get_frontend_corpus() -> list:

    global _FRONTEND_CORPUS

    if _FRONTEND_CORPUS is None:

        try:

            with open(_CORPUS_PATH) as f:

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _get_redis() -> Optional[redis_lib.Redis]:

    """Lazy Redis connection. Returns None if unavailable — never raises."""

    global _redis_client

    if _redis_client is not None:

        return _redis_client

    try:

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

"""

datanexus/tools/t10.py — T10 OSS Dependency & Vulnerability Intelligence.

Spec: DataNexus_MCP_Spec_v7_3.docx  Section 11.3 / Table 140  (authoritative)

Exactly 5 data functions + 2 infrastructure stubs = 7 total.

Data sources:

  Primary:   Google OSV.dev API (api.osv.dev/v1) — Apache 2.0, no key.

  Secondary: NIST NVD CVE API (services.nvd.nist.gov) — public domain, no key.

  Supporting: deps.dev API (api.deps.dev/v3alpha) — Apache 2.0, no key.

Hard stops (Section 12.5):

  - NEVER return

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _set_redis_client(client: Optional[redis_lib.Redis]) -> None:

    """Inject a Redis client (e.g. fakeredis) for testing. Pass None to reset."""

    global _redis_client

    _redis_client = client

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

"""

datanexus/tools/nonprofit_sprint7.py — Sprint 7 nonprofit depth tools.

Tools:

  search_nonprofits_by_category    — search US nonprofits by NTEE category + state

  fetch_nonprofit_financial_trends — multi-year revenue/expense/asset trends for a nonprofit

OQ1 resolved: ProPublica /api/v2/organizations/{ein}.json returns pre-computed

fields (totrevenue, totfuncexpns, totprgmrevnue, netassetsend, tax_prd_yr) in

filings_with_data. No raw 990 JSON parsing needed.

Circuit breaker: _propublica_br

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

"""

datanexus/tools/security_sprint6.py — Sprint 6 security tools.

Tools:

  fetch_package_maintainer_history  — maintainer health + anomaly score

  fetch_package_risk_brief          — SHIP/CAUTION/BLOCK aggregator

  detect_typosquatting              — Damerau-Levenshtein vs top-10k (added in final step)

All are thin MCP wrappers. Logic lives in _security_utils.py / _maintainer_utils.py.

HTTP self-calls are forbidden — utilities are called directly.

"""

import asyncio

import logging

import tim

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

connection is attempted rather than returning a broken client whose write errors

    would be silently swallowed by the outer try/except in report_feedback().

    """

    global _redis_client

    if _redis_client is not None:

        try:

            _redis_client.ping()

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _set_caller_id(caller_id: Optional[str]) -> None:

    """Inject a fixed caller_id for testing. Pass None to reset."""

    global _test_caller_id

    _test_caller_id = caller_id

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

"""

datanexus/tools/t04.py — T04 IRS 990 / Nonprofit Data tool.

Spec: DataNexus_MCP_Spec_v7_3.docx  Section 12.4 / Phase 2 Step B

Exactly 3 data functions + 2 infrastructure stubs = 5 total.

(Section 11.3, Table 163 — authoritative signatures)

Data sources:

  Primary:   IRS EO BMF (irs.gov) — public domain

  Secondary: IRS TEOS bulk downloads (irs.gov) — public domain

  Tertiary:  UK Charity Commission public bulk extract (no auth required)

             ccewuksprdoneregsadata1.blob.core.windo

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

Evidence

def _get_redis() -> Optional[redis_lib.Redis]:

    """Lazy Redis connection. Returns None if unavailable — never raises."""

    global _redis_client

    if _redis_client is not None:

        return _redis_client

    try:

Remediation

Partition shared state by caller identity, or eliminate it. Wrong: _cache = {} @mcp.tool() def search(query: str) -> list: if query in _cache: return _cache[query] result = _expensive(query) _cache[query] = result return result Right (key by caller): _cache: dict[str, dict] = {} @mcp.tool() def search(query: str, ctx) -> list: user_cache = _cache.setdefault(ctx.user_id, {}) if query in user_cache:

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

Remediation

Add a structured audit-event emit immediately after every destructive sink. Minimum schema: actor, action, target, outcome, request_id. Python: from acme.audit import audit_log @mcp.tool() def delete_record(token: str, record_id: str) -> dict: actor = verify_token(token) db.execute("DELETE FROM records WHERE id = %s", (record_id,)) audit_log( actor=actor.sub, action="delete_record", target=record_id, outcome=

Evidence

"""

datanexus/tools/api_key_sprint8a.py — Sprint 8A

Three API key management tools + _UsageMiddleware.

Tools (sub-server: datanexus-apikeys):

  generate_api_key  — issue a new dnx_... key, store email in DB

  rotate_api_key    — revoke current key, issue replacement

  revoke_api_key    — mark key revoked, invalidate Redis cache

_UsageMiddleware (FastMCP Middleware subclass):

  Intercepts every tool/call, increments per-user monthly Redis counter,

  injects usage fields into the response.  Fai

Remediation

Add a structured audit-event emit immediately after every destructive sink. Minimum schema: actor, action, target, outcome, request_id. Python: from acme.audit import audit_log @mcp.tool() def delete_record(token: str, record_id: str) -> dict: actor = verify_token(token) db.execute("DELETE FROM records WHERE id = %s", (record_id,)) audit_log( actor=actor.sub, action="delete_record", target=record_id, outcome=

Evidence

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY . .

RUN chmod +x start.sh

ENV PYTHONUNBUFFERED=1

# start.sh runs dashboard (8101) + MCP server (8000)

CMD ["./start.sh"]

Remediation

Create and switch to a non-root user before the CMD / ENTRYPOINT: RUN adduser --system --uid 1000 app USER 1000 Or reuse the base image's shipped non-root account (e.g. `USER nobody`, `USER nonroot` on distroless). Multi-stage builds only need the USER directive in the final stage.

Evidence

"""

    try:

        import jellyfish

        return jellyfish.damerau_levenshtein_distance(s1, s2)

    except ImportError:

        pass

    # Pure Python fallback

    len1, len2 = len(s1), len(s2)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

for val in results:

            if val is not None:

                try:

                    total += int(val)

                except (ValueError, TypeError):

                    pass

        return total

    except Exception as exc:

        log.warning("daily_digest: error reading call counters — %s", exc)

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

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:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

count += 1

        if count % batch_size == 0:

            try:

                pipe.execute()

            except Exception:

                pass

            pipe = r.pipeline(transaction=False)

    try:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.

Evidence

if count % batch_size == 0:

            try:

                pipe.execute()

            except Exception:

                pass

            pipe = r.pipeline(transaction=False)

    try:

Remediation

Log the exception at minimum (`logger.exception(e)`), emit a metric, or re-raise if the error is not recoverable. If you genuinely want to ignore an exception, say so with a comment.