Upload 14 files

Dockerfile

@@ -1,18 +1,23 @@
-FROM python:3.10
+FROM python:3.11-slim
 
-RUN useradd -m -u 1000 user && python -m pip install --upgrade pip
-USER user
-ENV PATH="/home/user/.local/bin:$PATH"
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+
+RUN useradd -m -u 1000 appuser
 
 WORKDIR /app
 
-COPY --chown=user ./requirements.txt requirements.txt
-RUN pip install --no-cache-dir --upgrade -r requirements.txt
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+
+COPY snap-python/ /app/snap-python/
+COPY app.py /app/app.py
 
-COPY --chown=user . /app
 ENV MCP_TRANSPORT=http
 ENV MCP_PORT=7860
 
 EXPOSE 7860
 
-CMD ["python", "snap-python/mcp_output/start_mcp.py"]
+USER appuser
+
+CMD ["python", "snap-python/mcp_output/start_mcp.py"]

README.md

@@ -1,10 +1,74 @@
 ---
-title: Snap Python
-emoji: 🌍
-colorFrom: red
-colorTo: pink
+title: snap-python MCP Service
+emoji: 🔧
+colorFrom: blue
+colorTo: indigo
 sdk: docker
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# snap-python MCP Service
+
+This deployment package wraps SNAP (Stanford Network Analysis Platform) Python functionality into a FastMCP service.
+
+## Available Tools
+
+- `adapter_health`
+- `list_modules`
+- `generate_graph`
+- `graph_info`
+- `basic_metrics`
+- `connectivity_metrics`
+- `degree_distribution`
+- `call_snap_function`
+
+Detailed parameter documentation and examples are in `snap-python/mcp_output/README_MCP.md`.
+
+## Local stdio usage (Claude Desktop / CLI)
+
+```bash
+cd snap-python/mcp_output
+python start_mcp.py
+```
+
+or:
+
+```bash
+cd snap-python/mcp_output/mcp_plugin
+python main.py
+```
+
+Default transport is `stdio`.
+
+## HTTP client usage
+
+```bash
+cd snap-python/mcp_output
+MCP_TRANSPORT=http MCP_PORT=8000 python start_mcp.py
+```
+
+Clients connect to:
+
+```text
+http://localhost:8000/mcp
+```
+
+## Docker / HuggingFace Spaces
+
+Docker entry point runs FastMCP directly:
+
+```text
+python snap-python/mcp_output/start_mcp.py
+```
+
+With environment:
+
+- `MCP_TRANSPORT=http`
+- `MCP_PORT=7860`
+
+Client endpoint:
+
+```text
+https://<your-space-host>/mcp
+```

app.py

@@ -1,45 +1,58 @@
-from fastapi import FastAPI
+from __future__ import annotations
+
 import os
 import sys
+from pathlib import Path
+from typing import Any
+
+from fastapi import FastAPI
 
-mcp_plugin_path = os.path.join(os.path.dirname(__file__), "snap-python", "mcp_output", "mcp_plugin")
-sys.path.insert(0, mcp_plugin_path)
 
-app = FastAPI(
-    title="Snap-Python MCP Service",
-    description="Auto-generated MCP service for snap-python",
-    version="1.0.0"
-)
+BASE_DIR = Path(__file__).resolve().parent
+PLUGIN_DIR = BASE_DIR / "snap-python" / "mcp_output" / "mcp_plugin"
+if str(PLUGIN_DIR) not in sys.path:
+    sys.path.insert(0, str(PLUGIN_DIR))
+
+app = FastAPI(title="snap-python MCP Info API", version="1.0.0")
+
 
 @app.get("/")
-def root():
+def root() -> dict[str, Any]:
     return {
-        "service": "Snap-Python MCP Service",
-        "version": "1.0.0",
-        "status": "running",
-        "transport": os.environ.get("MCP_TRANSPORT", "http")
+        "service": "snap-python MCP Service",
+        "mcp_transport": os.getenv("MCP_TRANSPORT", "stdio"),
+        "mcp_port": int(os.getenv("MCP_PORT", "8000")),
+        "info": "This FastAPI app is supplementary and does not run the MCP server.",
     }
 
+
 @app.get("/health")
-def health_check():
-    return {"status": "healthy", "service": "snap-python MCP"}
+def health() -> dict[str, str]:
+    return {"status": "healthy"}
+
 
 @app.get("/tools")
-def list_tools():
-    try:
-        from mcp_service import create_app
-        mcp_app = create_app()
-        tools = []
-        for tool_name, tool_func in mcp_app.tools.items():
-            tools.append({
-                "name": tool_name,
-                "description": tool_func.__doc__ or "No description available"
-            })
-        return {"tools": tools}
-    except Exception as e:
-        return {"error": f"Failed to load tools: {str(e)}"}
+def tools() -> dict[str, Any]:
+    from mcp_service import create_app
+
+    mcp = create_app()
+    tool_items: list[dict[str, Any]] = []
+
+    raw_tools = getattr(mcp, "tools", None)
+    if isinstance(raw_tools, dict):
+        for name, value in raw_tools.items():
+            description = getattr(value, "description", "")
+            tool_items.append({"name": str(name), "description": str(description)})
+    elif isinstance(raw_tools, list):
+        for value in raw_tools:
+            name = getattr(value, "name", "")
+            description = getattr(value, "description", "")
+            tool_items.append({"name": str(name), "description": str(description)})
+
+    return {"count": len(tool_items), "tools": tool_items}
+
 
 if __name__ == "__main__":
     import uvicorn
-    port = int(os.environ.get("PORT", 7860))
-    uvicorn.run(app, host="0.0.0.0", port=port)
+
+    uvicorn.run("app:app", host="0.0.0.0", port=int(os.getenv("PORT", "7860")), reload=False)

port.json

@@ -1,5 +1 @@
-{
-  "repo": "snap-python",
-  "port": 7916,
-  "timestamp": 1773243382
-}
+{"port": 7860}

requirements.txt

@@ -1,4 +1,4 @@
 fastmcp
+snap-stanford
 fastapi
-uvicorn[standard]
-pydantic>=2.0.0
+uvicorn

run_docker.ps1

@@ -1,26 +1,8 @@
-cd $PSScriptRoot
 $ErrorActionPreference = "Stop"
-$entryName = if ($env:MCP_ENTRY_NAME) { $env:MCP_ENTRY_NAME } else { "snap-python" }
-$entryUrl  = if ($env:MCP_ENTRY_URL)  { $env:MCP_ENTRY_URL  } else { "http://localhost:7916/mcp" }
-$imageName = if ($env:MCP_IMAGE_NAME) { $env:MCP_IMAGE_NAME } else { "snap-python-mcp" }
-$mcpDir = Join-Path $env:USERPROFILE ".cursor"
-$mcpPath = Join-Path $mcpDir "mcp.json"
-if (!(Test-Path $mcpDir)) { New-Item -ItemType Directory -Path $mcpDir | Out-Null }
-$config = @{}
-if (Test-Path $mcpPath) {
-  try { $config = Get-Content $mcpPath -Raw | ConvertFrom-Json } catch { $config = @{} }
-}
-$serversOrdered = [ordered]@{}
-if ($config -and ($config.PSObject.Properties.Name -contains "mcpServers") -and $config.mcpServers) {
-  $existing = $config.mcpServers
-  if ($existing -is [pscustomobject]) {
-    foreach ($p in $existing.PSObject.Properties) { if ($p.Name -ne $entryName) { $serversOrdered[$p.Name] = $p.Value } }
-  } elseif ($existing -is [System.Collections.IDictionary]) {
-    foreach ($k in $existing.Keys) { if ($k -ne $entryName) { $serversOrdered[$k] = $existing[$k] } }
-  }
-}
-$serversOrdered[$entryName] = @{ url = $entryUrl }
-$config = @{ mcpServers = $serversOrdered }
-$config | ConvertTo-Json -Depth 10 | Set-Content -Path $mcpPath -Encoding UTF8
+
+$portConfig = Get-Content -Raw -Path "port.json" | ConvertFrom-Json
+$port = [int]$portConfig.port
+$imageName = "snap-python-mcp"
+
 docker build -t $imageName .
-docker run --rm -p 7916:7860 $imageName
+docker run --rm -p "${port}:${port}" -e MCP_PORT=$port $imageName

run_docker.sh

@@ -1,75 +1,8 @@
 #!/usr/bin/env bash
 set -euo pipefail
-cd "$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
-mcp_entry_name="${MCP_ENTRY_NAME:-snap-python}"
-mcp_entry_url="${MCP_ENTRY_URL:-http://localhost:7916/mcp}"
-mcp_dir="${HOME}/.cursor"
-mcp_path="${mcp_dir}/mcp.json"
-mkdir -p "${mcp_dir}"
-if command -v python3 >/dev/null 2>&1; then
-python3 - "${mcp_path}" "${mcp_entry_name}" "${mcp_entry_url}" <<'PY'
-import json, os, sys
-path, name, url = sys.argv[1:4]
-cfg = {"mcpServers": {}}
-if os.path.exists(path):
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
-    except Exception:
-        cfg = {"mcpServers": {}}
-if not isinstance(cfg, dict):
-    cfg = {"mcpServers": {}}
-servers = cfg.get("mcpServers")
-if not isinstance(servers, dict):
-    servers = {}
-ordered = {}
-for k, v in servers.items():
-    if k != name:
-        ordered[k] = v
-ordered[name] = {"url": url}
-cfg = {"mcpServers": ordered}
-with open(path, "w", encoding="utf-8") as f:
-    json.dump(cfg, f, indent=2, ensure_ascii=False)
-PY
-elif command -v python >/dev/null 2>&1; then
-python - "${mcp_path}" "${mcp_entry_name}" "${mcp_entry_url}" <<'PY'
-import json, os, sys
-path, name, url = sys.argv[1:4]
-cfg = {"mcpServers": {}}
-if os.path.exists(path):
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
-    except Exception:
-        cfg = {"mcpServers": {}}
-if not isinstance(cfg, dict):
-    cfg = {"mcpServers": {}}
-servers = cfg.get("mcpServers")
-if not isinstance(servers, dict):
-    servers = {}
-ordered = {}
-for k, v in servers.items():
-    if k != name:
-        ordered[k] = v
-ordered[name] = {"url": url}
-cfg = {"mcpServers": ordered}
-with open(path, "w", encoding="utf-8") as f:
-    json.dump(cfg, f, indent=2, ensure_ascii=False)
-PY
-elif command -v jq >/dev/null 2>&1; then
-  name="${mcp_entry_name}"; url="${mcp_entry_url}"
-  if [ -f "${mcp_path}" ]; then
-    tmp="$(mktemp)"
-    jq --arg name "$name" --arg url "$url" '
-      .mcpServers = (.mcpServers // {})
-      | .mcpServers as $s
-      | ($s | with_entries(select(.key != $name))) as $base
-      | .mcpServers = ($base + {($name): {"url": $url}})
-    ' "${mcp_path}" > "${tmp}" && mv "${tmp}" "${mcp_path}"
-  else
-    printf '{ "mcpServers": { "%s": { "url": "%s" } } }
-' "$name" "$url" > "${mcp_path}"
-  fi
-fi
-docker build -t snap-python-mcp .
-docker run --rm -p 7916:7860 snap-python-mcp
+
+PORT=$(python3 -c "import json;print(json.load(open('port.json'))['port'])")
+IMAGE_NAME="snap-python-mcp"
+
+docker build -t "${IMAGE_NAME}" .
+docker run --rm -p "${PORT}:${PORT}" -e MCP_PORT="${PORT}" "${IMAGE_NAME}"

snap-python/mcp_output/README_MCP.md

@@ -1,110 +1,115 @@
-# snap-python MCP (Model Context Protocol) Service README
-
-## 1) Project Introduction
-
-This MCP (Model Context Protocol) service exposes graph analytics capabilities built on top of the `snap-python` repository (Stanford SNAP bindings + `snapx` Pythonic API).  
-It is designed for developer workflows where an LLM or client app needs to:
-
-- Create/load graphs
-- Run shortest path and connectivity analysis
-- Compute community/centrality metrics
-- Convert between Python data structures and graph objects
-
-Main usable layers:
-
-- `swig.snap` (high-performance SNAP bindings, broad algorithm coverage)
-- `snapx` (NetworkX-like API wrapping SNAP-style behavior)
-
----
-
-## 2) Installation Method
-
-### Requirements
-
-- Python 3.x
-- `pip`
-- Build toolchain for native extension (C/C++ compiler, SWIG-compatible environment) for `swig.snap` usage
-- Optional: `networkx`, `numpy`, `scipy` for interoperability or specific algorithm paths
-
-### Install
-
-- Install from repository root:
-  - `pip install .`
-- Editable/dev install:
-  - `pip install -e .`
-
-If native build fails, you can still use pure Python `snapx` portions where applicable, but full SNAP performance/features require successful extension build.
-
----
-
-## 3) Quick Start
-
-### Basic graph + shortest path (snapx-style)
-
-- Create graph with `snapx.classes.graph.Graph` or generators (`snapx.generators.classic.path_graph`)
-- Use unweighted/weighted shortest path modules:
-  - `snapx.algorithms.shortest_paths.unweighted.single_source_shortest_path`
-  - `snapx.algorithms.shortest_paths.weighted.dijkstra_path`
-
-### Core analysis functions (SNAP SWIG layer)
-
-Typical high-value functions available in `snap` include:
-
-- Connectivity: `GetWccs`, `GetSccs`, `IsConnected`, `GetMxWcc`
-- Path/BFS: `GetBfsTree`, `GetShortPath`, `GetBfsEffDiam`
-- Centrality: `GetDegreeCentr`, `GetClosenessCentr`, `GetBetweennessCentr`
-- Community: `CommunityCNM`, `CommunityGirvanNewman`
-- Generation: `GenRndGnm`, `GenPrefAttach`, `GenSmallWorld`, `GenRMat`
-
----
-
-## 4) Available Tools and Endpoints List
-
-Recommended MCP (Model Context Protocol) service surface (developer-oriented grouping):
-
-- `graph.create`
-  - Create empty/path/basic graphs (snapx generators, SNAP constructors)
-- `graph.load`
-  - Load edge lists / persisted graphs (`LoadEdgeList`, `*_Load`)
-- `graph.convert`
-  - Convert dict-of-dicts and Python structures (`to_snapx_graph`, `from_dict_of_dicts`)
-- `analysis.connectivity`
-  - Components and connectedness (`connected_components`, `GetWccs`, `GetSccs`)
-- `analysis.shortest_path`
-  - Unweighted + Dijkstra + all-pairs (`single_source_shortest_path`, `dijkstra_path`, `johnson`)
-- `analysis.centrality`
-  - Degree/closeness/betweenness/eigenvector variants
-- `analysis.community`
-  - CNM and Girvan-Newman community detection
-- `analysis.stats`
-  - Degree distributions, triads, clustering, diameter/ANF estimates
-- `graph.relabel`
-  - Node relabel operations (`relabel_nodes`, integer relabeling)
-
-Note: Repository does not define a ready-made CLI endpoint set; expose these as MCP (Model Context Protocol) service methods in your host runtime.
-
----
-
-## 5) Common Issues and Notes
-
-- Native build friction:
-  - Most installation issues come from missing compiler/SWIG toolchain.
-- Mixed API layers:
-  - `swig.snap` and `snapx` overlap conceptually; keep service contracts explicit about which backend each endpoint uses.
-- Performance:
-  - Prefer SWIG SNAP functions for large graphs and heavy analytics.
-- Compatibility:
-  - Some conversion paths may assume NetworkX-like inputs.
-- Testing/examples:
-  - Use `test/` and `examples/` as validation references for endpoint behavior and expected outputs.
-
----
-
-## 6) Reference Links / Documentation
-
-- Repository: https://github.com/snap-stanford/snap-python
-- Root docs: `README.md`, `troubleshooting.md`, `RELEASE.txt`
-- SWIG notes: `swig/README.txt`
-- Docker/environment help: `docker/README.md`
-- Examples: `examples/`, `dev/examples/`
-- Tests: `test/`, `dev/test/`
+# snap-python MCP Layer
+
+This folder contains the FastMCP adapter layer for exposing core SNAP capabilities as MCP tools.
+
+## Exposed MCP Tools
+
+1. **adapter_health**
+   - Parameters: none
+   - Returns adapter mode, module load counts, and source path diagnostics.
+
+2. **list_modules**
+   - Parameters: none
+   - Returns loaded modules and failed imports.
+
+3. **generate_graph**
+   - Parameters:
+     - `model: str` (`rnd_undirected`, `rnd_directed`, `pref_attach`, `small_world`, `rmat`, `forest_fire`)
+     - `num_nodes: int`
+     - `num_edges: int`
+     - `param: float = 0.1`
+     - `seed: int = 1`
+   - Returns an in-memory `graph_id` and summary stats.
+
+4. **graph_info**
+   - Parameters:
+     - `graph_id: str`
+   - Returns node/edge counts and empty-state flag.
+
+5. **basic_metrics**
+   - Parameters:
+     - `graph_id: str`
+     - `sample_nodes: int = 100`
+   - Returns triads, clustering coefficient, and approximate diameter.
+
+6. **connectivity_metrics**
+   - Parameters:
+     - `graph_id: str`
+     - `max_items: int = 20`
+   - Returns largest component fractions plus WCC/SCC distributions.
+
+7. **degree_distribution**
+   - Parameters:
+     - `graph_id: str`
+     - `max_items: int = 30`
+   - Returns sampled `(degree, count)` pairs.
+
+8. **call_snap_function**
+   - Parameters:
+     - `module_name: str`
+     - `function_name: str`
+     - `args_json: str = "[]"`
+     - `kwargs_json: str = "{}"`
+   - Generic dynamic executor for loaded modules.
+
+All tool responses follow:
+
+```json
+{
+  "success": true,
+  "result": {},
+  "error": null
+}
+```
+
+## Example Usage
+
+1. Generate a graph:
+
+```json
+{
+  "tool": "generate_graph",
+  "arguments": {
+    "model": "rmat",
+    "num_nodes": 10000,
+    "num_edges": 50000,
+    "param": 0.1,
+    "seed": 42
+  }
+}
+```
+
+2. Compute metrics:
+
+```json
+{
+  "tool": "basic_metrics",
+  "arguments": {
+    "graph_id": "<returned-graph-id>",
+    "sample_nodes": 200
+  }
+}
+```
+
+## Run Locally (stdio)
+
+```bash
+cd snap-python/mcp_output
+python start_mcp.py
+```
+
+or:
+
+```bash
+cd snap-python/mcp_output/mcp_plugin
+python main.py
+```
+
+## Run via HTTP Transport
+
+```bash
+cd snap-python/mcp_output
+MCP_TRANSPORT=http MCP_PORT=8000 python start_mcp.py
+```
+
+The MCP HTTP endpoint is served at `/mcp`.

snap-python/mcp_output/mcp_plugin/adapter.py

@@ -1,383 +1,180 @@
-import os
-import sys
+from __future__ import annotations
+
 import importlib
-import importlib.util
-from typing import Any, Dict, Optional, Tuple
+import pkgutil
+import sys
+from pathlib import Path
+from types import ModuleType
+from typing import Any
 
-source_path = os.path.join(
-    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
-    "source",
-)
-sys.path.insert(0, source_path)
 
+def _prepend_source_path() -> Path:
+    plugin_dir = Path(__file__).resolve().parent
+    source_dir = plugin_dir.parent.parent / "source"
+    if source_dir.exists() and str(source_dir) not in sys.path:
+        sys.path.insert(0, str(source_dir))
+    return source_dir
 
-class Adapter:
-    """
-    Import-mode adapter for snap-python repository integration.
 
-    This adapter attempts to import repository-local modules discovered by analysis and
-    exposes callable wrappers with unified response dictionaries.
+SOURCE_DIR = _prepend_source_path()
 
-    All public methods return:
-    {
-        "status": "success" | "error" | "fallback",
-        "mode": "import" | "fallback",
-        ...
-    }
-    """
 
-    # -------------------------------------------------------------------------
-    # Lifecycle and module management
-    # -------------------------------------------------------------------------
+class Adapter:
     def __init__(self) -> None:
-        """
-        Initialize adapter in import mode and attempt best-effort module loading.
-        """
-        self.mode = "import"
-        self._modules: Dict[str, Any] = {}
-        self._import_errors: Dict[str, str] = {}
-        self._load_modules()
-
-    def _result(self, status: str, **kwargs: Any) -> Dict[str, Any]:
-        data = {"status": status, "mode": self.mode}
-        data.update(kwargs)
-        return data
-
-    def _safe_import(self, key: str, module_path: str) -> None:
-        try:
-            self._modules[key] = importlib.import_module(module_path)
-        except Exception as e:
-            self._modules[key] = None
-            self._import_errors[key] = (
-                f"Failed to import '{module_path}'. "
-                f"Ensure repository source is available under '{source_path}' and dependencies are installed. "
-                f"Details: {e}"
-            )
-
-    def _load_hyphen_module(self, key: str, relative_path: str) -> None:
-        abs_path = os.path.join(source_path, relative_path)
-        try:
-            if not os.path.exists(abs_path):
-                raise FileNotFoundError(
-                    f"Module file not found at '{abs_path}'. Check repository extraction path."
-                )
-            spec = importlib.util.spec_from_file_location(key, abs_path)
-            if spec is None or spec.loader is None:
-                raise ImportError(
-                    f"Could not create import spec for '{abs_path}'."
-                )
-            module = importlib.util.module_from_spec(spec)
-            spec.loader.exec_module(module)
-            self._modules[key] = module
-        except Exception as e:
-            self._modules[key] = None
-            self._import_errors[key] = (
-                f"Failed to load module from '{abs_path}'. "
-                f"Verify file presence and Python compatibility. Details: {e}"
-            )
-
-    def _load_modules(self) -> None:
-        """
-        Load all discovered modules/classes/functions from analysis.
-        """
-        # Classes in setup modules
-        self._safe_import("setup_setup", "setup")
-        self._safe_import("swig_setup", "swig.setup")
-
-        # Functions in standard importable module
-        self._safe_import("genClassFnExt", "swig.gen.genClassFn.archive.genClassFnExt")
-
-        # Functions in modules with hyphenated filenames (manual load)
-        self._load_hyphen_module("disp_custom", os.path.join("swig", "gen", "disp-custom.py"))
-        self._load_hyphen_module("tneanet_cpp", os.path.join("dev", "examples", "tneanet-cpp.py"))
-        self._load_hyphen_module("snapswig_check", os.path.join("dev", "examples", "snapswig-check.py"))
-
-        if self._import_errors:
-            self.mode = "fallback"
-
-    def health_check(self) -> Dict[str, Any]:
-        """
-        Return adapter/module import status.
-
-        Returns:
-            dict: Unified status payload with loaded module keys and import errors.
-        """
-        loaded = [k for k, v in self._modules.items() if v is not None]
-        return self._result(
-            "success" if not self._import_errors else "fallback",
-            loaded_modules=loaded,
-            import_errors=self._import_errors,
-            guidance=(
-                "If imports failed, ensure the repository is present under the configured source path, "
-                "and install optional build/runtime dependencies for SWIG-based modules."
-            ),
-        )
-
-    # -------------------------------------------------------------------------
-    # Class instance creators
-    # -------------------------------------------------------------------------
-    def create_setup_swigbuild(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Create instance of setup.SwigBuild.
-
-        Parameters:
-            *args: Positional arguments forwarded to setup.SwigBuild constructor.
-            **kwargs: Keyword arguments forwarded to setup.SwigBuild constructor.
-
-        Returns:
-            dict: status and created instance or actionable error.
-        """
-        module = self._modules.get("setup_setup")
-        if module is None or not hasattr(module, "SwigBuild"):
-            return self._result(
-                "fallback",
-                error="Class 'SwigBuild' in module 'setup' is unavailable.",
-                guidance="Ensure 'setup.py' is importable and supports runtime class instantiation.",
-            )
-        try:
-            instance = module.SwigBuild(*args, **kwargs)
-            return self._result("success", class_name="SwigBuild", instance=instance)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"Failed to instantiate setup.SwigBuild: {e}",
-                guidance="Check constructor arguments required by distutils/setuptools command classes.",
-            )
-
-    def create_setup_swigextension(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Create instance of setup.SwigExtension.
-        """
-        module = self._modules.get("setup_setup")
-        if module is None or not hasattr(module, "SwigExtension"):
-            return self._result(
-                "fallback",
-                error="Class 'SwigExtension' in module 'setup' is unavailable.",
-                guidance="Ensure setup module loads successfully.",
-            )
-        try:
-            instance = module.SwigExtension(*args, **kwargs)
-            return self._result("success", class_name="SwigExtension", instance=instance)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"Failed to instantiate setup.SwigExtension: {e}",
-                guidance="Provide valid extension constructor parameters (name, sources, and config).",
-            )
-
-    def create_swig_setup_pkgbuild(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Create instance of swig.setup.PkgBuild.
-        """
-        module = self._modules.get("swig_setup")
-        if module is None or not hasattr(module, "PkgBuild"):
-            return self._result(
-                "fallback",
-                error="Class 'PkgBuild' in module 'swig.setup' is unavailable.",
-                guidance="Ensure swig setup module is present and importable.",
-            )
-        try:
-            instance = module.PkgBuild(*args, **kwargs)
-            return self._result("success", class_name="PkgBuild", instance=instance)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"Failed to instantiate swig.setup.PkgBuild: {e}",
-                guidance="Verify arguments expected by build command classes.",
-            )
-
-    def create_swig_setup_swigextension(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Create instance of swig.setup.SwigExtension.
-        """
-        module = self._modules.get("swig_setup")
-        if module is None or not hasattr(module, "SwigExtension"):
-            return self._result(
-                "fallback",
-                error="Class 'SwigExtension' in module 'swig.setup' is unavailable.",
-                guidance="Ensure swig setup module import succeeds.",
-            )
-        try:
-            instance = module.SwigExtension(*args, **kwargs)
-            return self._result("success", class_name="SwigExtension", instance=instance)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"Failed to instantiate swig.setup.SwigExtension: {e}",
-                guidance="Pass valid extension initialization arguments.",
-            )
-
-    # -------------------------------------------------------------------------
-    # Function call wrappers
-    # -------------------------------------------------------------------------
-    def call_convert_esubgraph(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call swig.gen.disp-custom.ConvertESubGraph.
-
-        Parameters:
-            *args: Forwarded to ConvertESubGraph.
-            **kwargs: Forwarded to ConvertESubGraph.
-        """
-        module = self._modules.get("disp_custom")
-        if module is None or not hasattr(module, "ConvertESubGraph"):
-            return self._result(
-                "fallback",
-                error="Function 'ConvertESubGraph' is unavailable.",
-                guidance="Check swig/gen/disp-custom.py availability and compatibility.",
-            )
-        try:
-            result = module.ConvertESubGraph(*args, **kwargs)
-            return self._result("success", function="ConvertESubGraph", result=result)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"ConvertESubGraph execution failed: {e}",
-                guidance="Validate argument types and graph object compatibility.",
-            )
-
-    def call_convert_graph(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call swig.gen.disp-custom.ConvertGraph.
-        """
-        module = self._modules.get("disp_custom")
-        if module is None or not hasattr(module, "ConvertGraph"):
-            return self._result(
-                "fallback",
-                error="Function 'ConvertGraph' is unavailable.",
-                guidance="Check swig/gen/disp-custom.py load status.",
-            )
-        try:
-            result = module.ConvertGraph(*args, **kwargs)
-            return self._result("success", function="ConvertGraph", result=result)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"ConvertGraph execution failed: {e}",
-                guidance="Ensure provided graph parameters match expected SNAP/SWIG types.",
-            )
-
-    def call_convert_subgraph(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call swig.gen.disp-custom.ConvertSubGraph.
-        """
-        module = self._modules.get("disp_custom")
-        if module is None or not hasattr(module, "ConvertSubGraph"):
-            return self._result(
-                "fallback",
-                error="Function 'ConvertSubGraph' is unavailable.",
-                guidance="Check swig/gen/disp-custom.py and repository source path.",
-            )
-        try:
-            result = module.ConvertSubGraph(*args, **kwargs)
-            return self._result("success", function="ConvertSubGraph", result=result)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"ConvertSubGraph execution failed: {e}",
-                guidance="Review node/edge subset arguments and graph input types.",
-            )
-
-    def call_gen_func_call(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call swig.gen.genClassFn.archive.genClassFnExt.genFuncCall.
-        """
-        module = self._modules.get("genClassFnExt")
-        if module is None or not hasattr(module, "genFuncCall"):
-            return self._result(
-                "fallback",
-                error="Function 'genFuncCall' is unavailable.",
-                guidance="Ensure module swig.gen.genClassFn.archive.genClassFnExt imports correctly.",
-            )
-        try:
-            result = module.genFuncCall(*args, **kwargs)
-            return self._result("success", function="genFuncCall", result=result)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"genFuncCall execution failed: {e}",
-                guidance="Verify expected inputs for class/function generation utility.",
-            )
-
-    def call_get_func_name(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call swig.gen.genClassFn.archive.genClassFnExt.getFuncName.
-        """
-        module = self._modules.get("genClassFnExt")
-        if module is None or not hasattr(module, "getFuncName"):
-            return self._result(
-                "fallback",
-                error="Function 'getFuncName' is unavailable.",
-                guidance="Check import status for swig.gen.genClassFn.archive.genClassFnExt.",
-            )
-        try:
-            result = module.getFuncName(*args, **kwargs)
-            return self._result("success", function="getFuncName", result=result)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"getFuncName execution failed: {e}",
-                guidance="Provide input matching expected parser/line format.",
-            )
-
-    def call_remove_first_param(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call swig.gen.genClassFn.archive.genClassFnExt.removeFirstParam.
-        """
-        module = self._modules.get("genClassFnExt")
-        if module is None or not hasattr(module, "removeFirstParam"):
-            return self._result(
-                "fallback",
-                error="Function 'removeFirstParam' is unavailable.",
-                guidance="Ensure genClassFnExt module loaded without syntax/runtime issues.",
-            )
-        try:
-            result = module.removeFirstParam(*args, **kwargs)
-            return self._result("success", function="removeFirstParam", result=result)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"removeFirstParam execution failed: {e}",
-                guidance="Check parameter string structure before invoking this utility.",
-            )
+        self.loaded_modules: dict[str, ModuleType] = {}
+        self.failed_modules: dict[str, str] = {}
+        self.mode = "ready"
+        self.source_dir = SOURCE_DIR
+        self._load_all_modules()
+
+    def _discover_candidate_modules(self) -> list[str]:
+        candidates = ["snap", "snapx"]
+        if self.source_dir.exists():
+            for module_info in pkgutil.iter_modules([str(self.source_dir)]):
+                name = module_info.name
+                if name.startswith("_"):
+                    continue
+                if name not in candidates:
+                    candidates.append(name)
+        return candidates
+
+    def _load_all_modules(self) -> None:
+        candidates = self._discover_candidate_modules()
+        for module_name in candidates:
+            try:
+                module = importlib.import_module(module_name)
+                self.loaded_modules[module_name] = module
+            except Exception as exc:
+                self.failed_modules[module_name] = str(exc)
+
+        if not self.loaded_modules:
+            self.mode = "blackbox"
+
+    def health(self) -> dict[str, Any]:
+        if self.loaded_modules:
+            return {
+                "status": "ok",
+                "mode": self.mode,
+                "loaded_count": len(self.loaded_modules),
+                "failed_count": len(self.failed_modules),
+                "source_dir": str(self.source_dir),
+            }
+        return {
+            "status": "fallback",
+            "mode": "blackbox",
+            "message": "No modules could be imported. Running in fallback mode.",
+            "loaded_count": 0,
+            "failed_count": len(self.failed_modules),
+            "source_dir": str(self.source_dir),
+        }
+
+    def list_modules(self) -> dict[str, Any]:
+        status = "ok" if self.loaded_modules else "fallback"
+        return {
+            "status": status,
+            "mode": self.mode,
+            "loaded": sorted(self.loaded_modules.keys()),
+            "failed": self.failed_modules,
+        }
+
+    def list_symbols(self, module_name: str, limit: int = 200) -> dict[str, Any]:
+        module = self.loaded_modules.get(module_name)
+        if module is None:
+            return {
+                "status": "error",
+                "message": f"Module '{module_name}' is not loaded",
+            }
+
+        names = [name for name in dir(module) if not name.startswith("__")]
+        return {
+            "status": "ok",
+            "module": module_name,
+            "count": len(names),
+            "symbols": names[: max(1, limit)],
+        }
+
+    def call_function(
+        self,
+        module_name: str,
+        function_name: str,
+        args: list[Any] | None = None,
+        kwargs: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        module = self.loaded_modules.get(module_name)
+        if module is None:
+            return {
+                "status": "error",
+                "message": f"Module '{module_name}' is not loaded",
+            }
+
+        fn = getattr(module, function_name, None)
+        if fn is None or not callable(fn):
+            return {
+                "status": "error",
+                "message": f"Function '{function_name}' not found in module '{module_name}'",
+            }
+
+        safe_args = args if args is not None else []
+        safe_kwargs = kwargs if kwargs is not None else {}
 
-    def call_tneanet_cpp_main(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call dev.examples.tneanet-cpp.main.
-        """
-        module = self._modules.get("tneanet_cpp")
-        if module is None or not hasattr(module, "main"):
-            return self._result(
-                "fallback",
-                error="Function 'main' in tneanet-cpp is unavailable.",
-                guidance="Check file dev/examples/tneanet-cpp.py exists and is valid.",
-            )
         try:
-            result = module.main(*args, **kwargs)
-            return self._result("success", function="tneanet_cpp.main", result=result)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"tneanet_cpp.main execution failed: {e}",
-                guidance="Confirm runtime prerequisites for example script are met.",
-            )
+            result = fn(*safe_args, **safe_kwargs)
+            return {
+                "status": "ok",
+                "module": module_name,
+                "function": function_name,
+                "result": result,
+            }
+        except Exception as exc:
+            return {
+                "status": "error",
+                "module": module_name,
+                "function": function_name,
+                "message": str(exc),
+            }
+
+    def create_instance(
+        self,
+        module_name: str,
+        class_name: str,
+        args: list[Any] | None = None,
+        kwargs: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        module = self.loaded_modules.get(module_name)
+        if module is None:
+            return {
+                "status": "error",
+                "message": f"Module '{module_name}' is not loaded",
+            }
+
+        cls = getattr(module, class_name, None)
+        if cls is None:
+            return {
+                "status": "error",
+                "message": f"Class '{class_name}' not found in module '{module_name}'",
+            }
+
+        if not isinstance(cls, type):
+            return {
+                "status": "error",
+                "message": f"Symbol '{class_name}' in module '{module_name}' is not a class",
+            }
+
+        safe_args = args if args is not None else []
+        safe_kwargs = kwargs if kwargs is not None else {}
 
-    def call_snapswig_check_main(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call dev.examples.snapswig-check.main.
-        """
-        module = self._modules.get("snapswig_check")
-        if module is None or not hasattr(module, "main"):
-            return self._result(
-                "fallback",
-                error="Function 'main' in snapswig-check is unavailable.",
-                guidance="Check file dev/examples/snapswig-check.py exists and imports succeed.",
-            )
         try:
-            result = module.main(*args, **kwargs)
-            return self._result("success", function="snapswig_check.main", result=result)
-        except Exception as e:
-            return self._result(
-                "error",
-                error=f"snapswig_check.main execution failed: {e}",
-                guidance="Ensure SWIG SNAP extension is built and importable before running checks.",
-            )
+            instance = cls(*safe_args, **safe_kwargs)
+            return {
+                "status": "ok",
+                "module": module_name,
+                "class": class_name,
+                "instance_type": type(instance).__name__,
+            }
+        except Exception as exc:
+            return {
+                "status": "error",
+                "module": module_name,
+                "class": class_name,
+                "message": str(exc),
+            }

snap-python/mcp_output/mcp_plugin/main.py

@@ -1,13 +1,11 @@
-"""
-MCP Service Auto-Wrapper - Auto-generated
-"""
 from mcp_service import create_app
 
-def main():
-    """Main entry point"""
+
+def main() -> None:
+    # Local stdio entry point only (Claude Desktop / CLI), not for Docker/web deployment.
     app = create_app()
-    return app
+    app.run(transport="stdio")
+
 
 if __name__ == "__main__":
-    app = main()
-    app.run()
+    main()

snap-python/mcp_output/mcp_plugin/mcp_service.py

@@ -1,271 +1,340 @@
+from __future__ import annotations
+
+import json
 import os
 import sys
-
-source_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "source")
-if source_path not in sys.path:
-    sys.path.insert(0, source_path)
+import uuid
+from pathlib import Path
+from typing import Any
 
 from fastmcp import FastMCP
 
-from setup import SwigExtension, SwigBuild
-from swig.setup import SwigExtension, PkgBuild
-from swig.gen.disp-custom import ConvertGraph, ConvertSubGraph, ConvertESubGraph
-from swig.gen.genClassFn.archive.genClassFnExt import removeFirstParam, genFuncCall, getFuncName
-from dev.examples.tneanet-cpp import main
-from dev.examples.snapswig-check import main
+from adapter import Adapter
 
-mcp = FastMCP("unknown_service")
 
+def _prepend_source_path() -> Path:
+    plugin_dir = Path(__file__).resolve().parent
+    source_dir = plugin_dir.parent.parent / "source"
+    if source_dir.exists() and str(source_dir) not in sys.path:
+        sys.path.insert(0, str(source_dir))
+    return source_dir
 
-@mcp.tool(name="swigbuild", description="SwigBuild class")
-def swigbuild(*args, **kwargs):
-    """SwigBuild class"""
-    try:
-        if SwigBuild is None:
-            return {"success": False, "result": None, "error": "Class SwigBuild is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = SwigBuild(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="swigextension", description="SwigExtension class")
-def swigextension(*args, **kwargs):
-    """SwigExtension class"""
-    try:
-        if SwigExtension is None:
-            return {"success": False, "result": None, "error": "Class SwigExtension is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = SwigExtension(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="pkgbuild", description="PkgBuild class")
-def pkgbuild(*args, **kwargs):
-    """PkgBuild class"""
-    try:
-        if PkgBuild is None:
-            return {"success": False, "result": None, "error": "Class PkgBuild is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = PkgBuild(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="swigextension", description="SwigExtension class")
-def swigextension(*args, **kwargs):
-    """SwigExtension class"""
+
+SOURCE_DIR = _prepend_source_path()
+
+mcp = FastMCP("snap-python-mcp")
+adapter = Adapter()
+
+GRAPH_STORE: dict[str, Any] = {}
+
+
+def _ok(result: Any) -> dict[str, Any]:
+    return {"success": True, "result": result, "error": None}
+
+
+def _err(message: str) -> dict[str, Any]:
+    return {"success": False, "result": None, "error": message}
+
+
+def _get_snap_module() -> Any:
+    snap_module = adapter.loaded_modules.get("snap")
+    if snap_module is not None:
+        return snap_module
     try:
-        if SwigExtension is None:
-            return {"success": False, "result": None, "error": "Class SwigExtension is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
-        
-        instance = SwigExtension(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="ConvertESubGraph", description="Auto-wrapped function ConvertESubGraph")
-def ConvertESubGraph(payload: dict):
+        import snap as snap_module  # type: ignore
+
+        return snap_module
+    except Exception as exc:
+        raise RuntimeError(f"Failed to import 'snap': {exc}") from exc
+
+
+def _get_graph(graph_id: str) -> Any:
+    graph = GRAPH_STORE.get(graph_id)
+    if graph is None:
+        raise KeyError(f"Graph id '{graph_id}' not found")
+    return graph
+
+
+def _int_pair_vector_to_list(vector_obj: Any, max_items: int) -> list[dict[str, int]]:
+    items: list[dict[str, int]] = []
+    length = int(getattr(vector_obj, "Len")())
+    for index in range(min(length, max_items)):
+        pair = vector_obj.GetI(index)
+        value1 = int(pair.GetVal1())
+        value2 = int(pair.GetVal2())
+        items.append({"x": value1, "count": value2})
+    return items
+
+
+@mcp.tool(name="adapter_health", description="Get adapter and module loading health status")
+def adapter_health() -> dict[str, Any]:
+    """Return adapter health.
+
+    Returns:
+        A standardized MCP response with adapter mode, loaded module counts,
+        and source path diagnostics.
+    """
     try:
-        if ConvertESubGraph is None:
-            return {"success": False, "result": None, "error": "Function ConvertESubGraph is not available"}
-        result = ConvertESubGraph(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="ConvertGraph", description="Auto-wrapped function ConvertGraph")
-def ConvertGraph(payload: dict):
+        result = adapter.health()
+        return _ok(result)
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="list_modules", description="List successfully loaded and failed modules")
+def list_modules() -> dict[str, Any]:
+    """List loaded modules.
+
+    Returns:
+        A standardized MCP response containing loaded and failed module maps.
+    """
     try:
-        if ConvertGraph is None:
-            return {"success": False, "result": None, "error": "Function ConvertGraph is not available"}
-        result = ConvertGraph(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="ConvertSubGraph", description="Auto-wrapped function ConvertSubGraph")
-def ConvertSubGraph(payload: dict):
+        return _ok(adapter.list_modules())
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="generate_graph", description="Generate a SNAP graph and keep it in an in-memory store")
+def generate_graph(
+    model: str,
+    num_nodes: int,
+    num_edges: int,
+    param: float = 0.1,
+    seed: int = 1,
+) -> dict[str, Any]:
+    """Generate a graph with SNAP and return a graph id.
+
+    Args:
+        model: Graph model name. One of: rnd_undirected, rnd_directed,
+            pref_attach, small_world, rmat, forest_fire.
+        num_nodes: Number of nodes to generate.
+        num_edges: Number of edges for models that require it.
+        param: Model-specific parameter. For `pref_attach`, this is node out-degree.
+            For `small_world`, this is rewiring probability.
+            For `forest_fire`, this is forward burn probability.
+        seed: Random seed for stochastic graph generation.
+
+    Returns:
+        A standardized MCP response with graph id and graph summary.
+    """
     try:
-        if ConvertSubGraph is None:
-            return {"success": False, "result": None, "error": "Function ConvertSubGraph is not available"}
-        result = ConvertSubGraph(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="genFuncCall", description="Auto-wrapped function genFuncCall")
-def genFuncCall(payload: dict):
+        if num_nodes <= 0:
+            return _err("num_nodes must be > 0")
+
+        snap = _get_snap_module()
+        graph = None
+        normalized = model.strip().lower()
+
+        if normalized == "rnd_undirected":
+            if num_edges <= 0:
+                return _err("num_edges must be > 0 for rnd_undirected")
+            graph = snap.GenRndGnm(snap.PUNGraph, int(num_nodes), int(num_edges), int(seed))
+        elif normalized == "rnd_directed":
+            if num_edges <= 0:
+                return _err("num_edges must be > 0 for rnd_directed")
+            graph = snap.GenRndGnm(snap.PNGraph, int(num_nodes), int(num_edges), int(seed))
+        elif normalized == "pref_attach":
+            node_out_degree = max(1, int(param))
+            graph = snap.GenPrefAttach(int(num_nodes), int(node_out_degree))
+        elif normalized == "small_world":
+            node_out_degree = max(2, int(num_edges / max(1, num_nodes)))
+            rewire_prob = min(max(float(param), 0.0), 1.0)
+            graph = snap.GenSmallWorld(int(num_nodes), int(node_out_degree), rewire_prob)
+        elif normalized == "rmat":
+            if num_edges <= 0:
+                return _err("num_edges must be > 0 for rmat")
+            rnd = snap.TRnd(int(seed))
+            rnd.Randomize()
+            graph = snap.GenRMat(int(num_nodes), int(num_edges), 0.45, 0.15, 0.15, rnd)
+        elif normalized == "forest_fire":
+            burn_prob = min(max(float(param), 0.01), 0.99)
+            graph = snap.GenForestFire(int(num_nodes), burn_prob, burn_prob)
+        else:
+            return _err(
+                "Unsupported model. Use one of: rnd_undirected, rnd_directed, "
+                "pref_attach, small_world, rmat, forest_fire"
+            )
+
+        graph_id = str(uuid.uuid4())
+        GRAPH_STORE[graph_id] = graph
+        return _ok(
+            {
+                "graph_id": graph_id,
+                "model": normalized,
+                "nodes": int(graph.GetNodes()),
+                "edges": int(graph.GetEdges()),
+            }
+        )
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="graph_info", description="Return node and edge counts for a stored graph")
+def graph_info(graph_id: str) -> dict[str, Any]:
+    """Get basic graph summary.
+
+    Args:
+        graph_id: Identifier returned by `generate_graph`.
+
+    Returns:
+        A standardized MCP response with node/edge statistics.
+    """
     try:
-        if genFuncCall is None:
-            return {"success": False, "result": None, "error": "Function genFuncCall is not available"}
-        result = genFuncCall(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="getFuncName", description="Auto-wrapped function getFuncName")
-def getFuncName(payload: dict):
+        graph = _get_graph(graph_id)
+        result = {
+            "graph_id": graph_id,
+            "nodes": int(graph.GetNodes()),
+            "edges": int(graph.GetEdges()),
+            "is_empty": bool(graph.Empty()),
+        }
+        return _ok(result)
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="basic_metrics", description="Compute triads, clustering coefficient, and approximate diameter")
+def basic_metrics(graph_id: str, sample_nodes: int = 100) -> dict[str, Any]:
+    """Compute baseline topology metrics for a graph.
+
+    Args:
+        graph_id: Identifier returned by `generate_graph`.
+        sample_nodes: Number of BFS samples used for approximate diameter.
+
+    Returns:
+        A standardized MCP response with triads, clustering, and diameter.
+    """
     try:
-        if getFuncName is None:
-            return {"success": False, "result": None, "error": "Function getFuncName is not available"}
-        result = getFuncName(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="removeFirstParam", description="Auto-wrapped function removeFirstParam")
-def removeFirstParam(payload: dict):
+        snap = _get_snap_module()
+        graph = _get_graph(graph_id)
+        triads = int(snap.GetTriads(graph))
+        clust_cf = float(snap.GetClustCf(graph))
+        diameter = int(snap.GetBfsFullDiam(graph, int(max(1, sample_nodes))))
+        return _ok(
+            {
+                "graph_id": graph_id,
+                "triads": triads,
+                "clustering_coefficient": clust_cf,
+                "approx_diameter": diameter,
+            }
+        )
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="connectivity_metrics", description="Compute largest WCC/SCC and their size distributions")
+def connectivity_metrics(graph_id: str, max_items: int = 20) -> dict[str, Any]:
+    """Compute connectivity-related metrics.
+
+    Args:
+        graph_id: Identifier returned by `generate_graph`.
+        max_items: Maximum number of distribution entries to return.
+
+    Returns:
+        A standardized MCP response with WCC/SCC summary and distributions.
+    """
     try:
-        if removeFirstParam is None:
-            return {"success": False, "result": None, "error": "Function removeFirstParam is not available"}
-        result = removeFirstParam(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="main", description="Auto-wrapped function main")
-def main(payload: dict):
+        snap = _get_snap_module()
+        graph = _get_graph(graph_id)
+
+        wcc_cnt = snap.TIntPrV()
+        scc_cnt = snap.TIntPrV()
+        snap.GetWccSzCnt(graph, wcc_cnt)
+        snap.GetSccSzCnt(graph, scc_cnt)
+
+        result = {
+            "graph_id": graph_id,
+            "max_wcc_fraction": float(snap.GetMxWccSz(graph)),
+            "max_scc_fraction": float(snap.GetMxSccSz(graph)),
+            "wcc_distribution": _int_pair_vector_to_list(wcc_cnt, max(1, max_items)),
+            "scc_distribution": _int_pair_vector_to_list(scc_cnt, max(1, max_items)),
+        }
+        return _ok(result)
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="degree_distribution", description="Return out-degree distribution as (degree, count) pairs")
+def degree_distribution(graph_id: str, max_items: int = 30) -> dict[str, Any]:
+    """Compute degree distribution for a stored graph.
+
+    Args:
+        graph_id: Identifier returned by `generate_graph`.
+        max_items: Maximum number of (degree, count) pairs to return.
+
+    Returns:
+        A standardized MCP response with degree histogram samples.
+    """
     try:
-        if main is None:
-            return {"success": False, "result": None, "error": "Function main is not available"}
-        result = main(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-@mcp.tool(name="main", description="Auto-wrapped function main")
-def main(payload: dict):
+        snap = _get_snap_module()
+        graph = _get_graph(graph_id)
+        deg_cnt = snap.TIntPrV()
+        snap.GetOutDegCnt(graph, deg_cnt)
+        return _ok(
+            {
+                "graph_id": graph_id,
+                "distribution": _int_pair_vector_to_list(deg_cnt, max(1, max_items)),
+            }
+        )
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="call_snap_function", description="Call a selected function from a loaded module using JSON arguments")
+def call_snap_function(
+    module_name: str,
+    function_name: str,
+    args_json: str = "[]",
+    kwargs_json: str = "{}",
+) -> dict[str, Any]:
+    """Call a dynamically selected function from loaded modules.
+
+    Args:
+        module_name: Module name (for example, `snap`).
+        function_name: Callable symbol name in the module.
+        args_json: JSON array string for positional arguments.
+        kwargs_json: JSON object string for keyword arguments.
+
+    Returns:
+        A standardized MCP response with dynamic call status and result metadata.
+    """
     try:
-        if main is None:
-            return {"success": False, "result": None, "error": "Function main is not available"}
-        result = main(**payload)
-        return {"success": True, "result": result, "error": None}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        parsed_args = json.loads(args_json)
+        parsed_kwargs = json.loads(kwargs_json)
+        if not isinstance(parsed_args, list):
+            return _err("args_json must decode to a JSON array")
+        if not isinstance(parsed_kwargs, dict):
+            return _err("kwargs_json must decode to a JSON object")
 
+        response = adapter.call_function(
+            module_name=module_name,
+            function_name=function_name,
+            args=parsed_args,
+            kwargs=parsed_kwargs,
+        )
+        if response.get("status") == "ok":
+            raw = response.get("result")
+            safe_result = raw if isinstance(raw, (str, int, float, bool, list, dict, type(None))) else str(raw)
+            return _ok(
+                {
+                    "status": "ok",
+                    "module": module_name,
+                    "function": function_name,
+                    "result": safe_result,
+                }
+            )
+        return _err(response.get("message", "Dynamic call failed"))
+    except Exception as exc:
+        return _err(str(exc))
 
 
-def create_app():
-    """Create and return FastMCP application instance"""
+def create_app() -> FastMCP:
     return mcp
 
+
 if __name__ == "__main__":
-    mcp.run(transport="http", host="0.0.0.0", port=8000)
+    transport = os.getenv("MCP_TRANSPORT", "stdio").strip().lower() or "stdio"
+    if transport == "http":
+        port = int(os.getenv("MCP_PORT", "8000"))
+        mcp.run(transport="http", host="0.0.0.0", port=port)
+    else:
+        mcp.run(transport="stdio")

snap-python/mcp_output/requirements.txt

@@ -1,5 +1,2 @@
 fastmcp
-fastapi
-uvicorn[standard]
-pydantic>=2.0.0
-snapx package modules under source.snapx.snapx
+snap-stanford

snap-python/mcp_output/start_mcp.py

@@ -1,30 +1,28 @@
+from __future__ import annotations
 
-"""
-MCP Service Startup Entry
-"""
-import sys
 import os
+import sys
+from pathlib import Path
+
 
-project_root = os.path.dirname(os.path.abspath(__file__))
-mcp_plugin_dir = os.path.join(project_root, "mcp_plugin")
-if mcp_plugin_dir not in sys.path:
-    sys.path.insert(0, mcp_plugin_dir)
+CURRENT_DIR = Path(__file__).resolve().parent
+PLUGIN_DIR = CURRENT_DIR / "mcp_plugin"
+if str(PLUGIN_DIR) not in sys.path:
+    sys.path.insert(0, str(PLUGIN_DIR))
 
 from mcp_service import create_app
 
-def main():
-    """Start FastMCP service"""
+
+def main() -> None:
+    transport = os.getenv("MCP_TRANSPORT", "stdio").strip().lower() or "stdio"
+    port = int(os.getenv("MCP_PORT", "8000"))
     app = create_app()
-    # Use environment variable to configure port, default 8000
-    port = int(os.environ.get("MCP_PORT", "8000"))
-    
-    # Choose transport mode based on environment variable
-    transport = os.environ.get("MCP_TRANSPORT", "stdio")
+
     if transport == "http":
         app.run(transport="http", host="0.0.0.0", port=port)
     else:
-        # Default to STDIO mode
-        app.run()
+        app.run(transport="stdio")
+
 
 if __name__ == "__main__":
-    main()
+    main()