maltego_mcp

An MCP server that turns an LLM into a Maltego CE investigation copilot — an AI-assisted OSINT platform built on Maltego Community Edition's native graph format.

Maltego CE has no live external API. Instead of trying to remote-control the desktop UI, this server uses Maltego's native .mtgx graph file (a ZIP archive of GraphML XML) as the integration surface:

1. The LLM builds / reads / edits an investigation graph held in memory.

2. Transforms — and high-level investigation workflows / machines — expand the graph (e.g. a domain → its IP addresses → open ports).

3. The LLM analyzes the graph (summaries, pivots, next steps), lays it out, and generates a shareable report.

4. The graph is saved as a .mtgx file that you open (or refresh) in Maltego CE.

A pluggable transform-provider layer keeps the design extensible: the built-in local provider needs no API keys, and external OSINT providers (VirusTotal, Shodan, SecurityTrails, Censys, Hunter.io, HaveIBeenPwned) activate automatically when their API keys are present — all without changing the graph core or the MCP tools.

Capabilities at a glance

Install as a Claude Code plugin (recommended)

This repo is both a plugin and a plugin marketplace. Install it straight from GitHub:

# in Claude Code
/plugin marketplace add SulimanAbdulrazzaq/maltego-mcp
/plugin install maltego-mcp@maltego-mcp

The bundled MCP server is launched with uv via uvx, which auto-installs the Python dependencies in an isolated environment — no manual pip install needed. Install uv once if you don't have it:

# Windows (PowerShell)
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

After install, the maltego_* tools are available in Claude. Set OSINT keys (optional) in your shell before starting Claude Code to enable those providers (see Transforms by provider).

Prefer not to use uv? See Manual / pip install and point the MCP command at the venv's maltego-mcp executable instead.

Windows troubleshooting (first launch)

On Windows, the very first launch occasionally fails with failed to rename ... Access is denied (os error 5) while unpacking pywin32 (a transitive dependency of mcp). This is Windows Defender briefly locking the files during real-time scanning — not a bug in the plugin. Fixes:

• Just retry — toggle the plugin off/on or restart Claude Code; it succeeds once Defender finishes scanning and the dependency is cached (verified: it works on the second attempt).

• Or pre-warm the cache once from a terminal: uvx --from <path-to-this-repo> maltego-mcp (Ctrl-C after it prints nothing — it's a stdio server), then retry in Claude Code.

• Or add an exclusion for %LOCALAPPDATA%\uv\cache in Windows Security → Virus & threat protection → Exclusions.

Related MCP server: CutterMCP-plus

Manual / pip install

cd maltego-mcp
python -m venv .venv
.venv/Scripts/activate        # Windows
# source .venv/bin/activate   # macOS/Linux
pip install -e .

Then register the stdio server with any MCP client, e.g.:

{
  "mcpServers": {
    "maltego": { "command": ".../.venv/Scripts/maltego-mcp.exe" }
  }
}

Running

The server speaks MCP over stdio (ideal for local desktop integration):

maltego-mcp
# or
python -m maltego_mcp.server

Register with an MCP client

Example mcp.json / client config entry:

{
  "mcpServers": {
    "maltego": {
      "command": "C:/Users/you/Desktop/maltego_mcp/.venv/Scripts/maltego-mcp.exe"
    }
  }
}

(Use the .venv Python on macOS/Linux: "command": ".../.venv/bin/maltego-mcp".)

Test with the MCP Inspector

npx @modelcontextprotocol/inspector maltego-mcp

Tools

Graph management

Entities & links

Transforms

High-level investigation workflows

Unified entry point (primary interface for agents)

AI-oriented analysis (the "copilot")

Investigation Memory (procedural memory)

Risk & confidence scoring

Real-time mode (optional)

Layout, CSV, import/export, link & graph management

Outcome-based learning (opt-in)

Learning is off by default (keeps results deterministic). Enable with MALTEGO_MCP_LEARNING=1 (or MALTEGO_MCP_LEARNING_PATH=/path/to/learning.json). When on, the server records per-transform outcomes across investigations and lets that history nudge maltego_next_best_actions.

Transforms by provider

Built-in local provider — no API key required:

• dns.domain_to_ip — domain/DNS name → IPv4 addresses (DNS A record) [network]

• dns.ip_to_host — IPv4 address → hostname (reverse DNS / PTR) [network]

• parse.url_to_domain — URL → domain (offline)

• parse.email_to_domain — email → domain (offline)

• parse.domain_to_website — domain → Website entity (offline)

External OSINT providers — activate when their env-var key is set (see maltego_list_providers). Transforms are always listed but only run when configured; a missing key yields an actionable message rather than an error.

Configure keys via environment variables (in your MCP client's env block or the shell), then restart the server. Example client config:

{
  "mcpServers": {
    "maltego": {
      "command": ".../.venv/Scripts/maltego-mcp.exe",
      "env": { "VIRUSTOTAL_API_KEY": "...", "SHODAN_API_KEY": "..." }
    }
  }
}

Example workflows

Manual (low-level):

maltego_create_graph(name="acme-recon")
maltego_add_entity(type="maltego.Domain", value="example.com")        # -> n0
maltego_run_transform(transform_name="dns.domain_to_ip", entity_id="n0")
maltego_apply_layout(algorithm="hierarchical")
maltego_save_graph(path="C:/Users/you/Desktop/acme-recon.mtgx")

One-call copilot (recommended for agents):

maltego_investigate(query="bob@example.com")   # detect → build → expand → analyze → recommend
maltego_explain_why(entity_id="n1")            # why is this entity here?
maltego_next_best_actions()                    # explainable, memory-aware ranking
maltego_rank_entities()                        # focus on the highest-priority findings
maltego_get_investigation_timeline()           # the reasoning trace
maltego_generate_report(format="html")
maltego_save_graph(path="C:/cases/example.mtgx")   # memory travels inside the .mtgx

AI-assisted (step-by-step):

maltego_investigate_domain(value="example.com")     # auto-runs transforms
maltego_identify_pivots()                            # find key nodes
maltego_next_best_actions()                          # what to do next (decision engine)
maltego_apply_layout(algorithm="radial")
maltego_generate_report(format="markdown")           # shareable report
maltego_save_graph(path="C:/Users/you/Desktop/example.mtgx")

Continue a previous investigation:

maltego_load_graph(path="C:/cases/old.mtgx")         # reopen
maltego_run_machine(machine_name="infrastructure_mapping", seed_value="example.com")
maltego_import_graph(path="C:/cases/related.mtgx")   # merge in another case

Bulk import from CSV:

maltego_import_csv(content="type,value,link_to\nDomain,example.com,\nIPv4Address,1.2.3.4,example.com\n")

Extending

Add a new OSINT provider

1. Create a module under src/maltego_mcp/transforms/osint/.

2. Write pure parser functions (dict -> list[ResultEntity]) and async run functions that read the API key from an env var via require_keys(...).

3. Register a ProviderInfo with providers.register(...) and your Transform(...) objects (with api_key_env=...) via registry.register(...).

4. Import the module from transforms/osint/__init__.py.

No changes to the graph core, orchestration, machines, or MCP tools are needed — new transforms automatically participate in investigate_*, machines, analysis, and suggestions.

Add an investigation machine

from maltego_mcp.machines import Machine, register_machine
register_machine(Machine(
    name="my_workflow", display_name="My Workflow", description="...",
    seed_type="maltego.Domain",
    transform_names=["dns.domain_to_ip", "vt.domain_to_subdomains"],
    allow_network=True, max_rounds=2,
))

Architecture

src/maltego_mcp/
├── server.py          # FastMCP server + all 49 tool registrations
├── models.py          # Pydantic input models
├── entities.py        # Maltego entity-type catalog
├── formatting.py      # markdown/JSON response helpers + error mapping
├── detect.py          # query -> entity type/value/machine (for maltego_investigate)
├── orchestration.py   # breadth-first engine + run_and_record (memory+events choke-point)
├── machines.py        # reusable workflow templates + registry
├── analysis.py        # deterministic summarize/explain/pivots/next-steps
├── recommendation.py  # Next Best Action decision engine (memory-aware, explainable)
├── scoring.py         # Risk & confidence engine (deterministic per-entity metrics)
├── memory.py          # Investigation Memory (procedural memory; storage + queries)
├── learning.py        # opt-in cross-investigation outcome learning (feeds NBA)
├── events.py          # architecture-agnostic event bus (real-time mode)
├── layout.py          # hierarchical / radial / force-directed layouts
├── csv_import.py      # CSV -> entities/links (type aliases, dedupe)
├── reporting.py       # deterministic Markdown / HTML reports (incl. quality scores)
├── graph/
│   ├── graph_store.py # in-memory Graph (+ .memory, merge, analysis helpers)
│   ├── mtgx_writer.py # Graph -> .mtgx (GraphML + positions + memory sidecar + zip)
│   └── mtgx_reader.py # .mtgx -> Graph (recovers positions + memory sidecar)
└── transforms/
    ├── base.py        # Transform/registry + ProviderInfo/ProviderRegistry (+ reliability)
    ├── local.py       # built-in no-auth transforms
    └── osint/         # external providers (VT, Shodan, SecurityTrails, ...)
        ├── base_http.py
        └── virustotal.py, shodan.py, securitytrails.py, censys.py, hunterio.py, hibp.py

Investigation Memory & determinism

• Procedural memory (memory.py) records every transform execution — the trigger entity, the chosen transform, why it was chosen, what it discovered, status, importance, and whether to reconsider it. It is stored on Graph.memory, kept separate from the graph structure, and serialized to a sidecar member (maltego_mcp/investigation_memory.json) inside the .mtgx — so it travels with the investigation but never affects Maltego CE compatibility (Maltego ignores unknown archive members).

• orchestration.run_and_record is the single choke-point through which the engine and the manual maltego_run_transform tool execute transforms, so memory and events are captured consistently everywhere.

• Scores (scoring.py) are computed deterministically from the graph + memory and provider reliability (ProviderInfo.reliability), so the same investigation always yields the same scores, recommendations, and reports.

Verifying a graph opens in Maltego CE

The .mtgx format is validated by our reader/writer round-trip and by checking entity property field names against Maltego's real definitions (e.g. Domain/Website → fqdn, Company/Organization → title, IPv4Address → ipv4-address). To confirm end-to-end in the actual app:

1. Install Maltego CE (free; requires a download + account at maltego.com).

2. Generate a sample: maltego_investigate("example.com") (or add a few entities) then maltego_save_graph(path="…/sample.mtgx"). A ready-made sample-verification.mtgx is produced on the Desktop by the test fixtures.

3. In Maltego CE: File → Import → Import Graph (or File → Open) and select the .mtgx.

4. Confirm entities render with their values populated, correct types, and links with labels. If a given type's value is blank, its main_property in src/maltego_mcp/entities.py needs correcting against Maltego's field id for that entity.

Status: the common entity types (Domain, IP, Email, Website, Person, Company, …) have had their field names verified against a reference; opening in a real Maltego CE install is the final confirmation step and is left to the user (no Maltego install was available here).

Notes & limitations

• Targets Maltego CE's file format; it does not remote-control the running desktop app. Re-open or refresh the .mtgx in Maltego after saving.

• The entity-type catalog is a curated subset; custom maltego.* types are accepted and saved as-is.

• Graphs live in process memory until saved; restarting the server clears unsaved graphs.

• Layout positions are written into the .mtgx as yFiles node graphics. Maltego CE may re-run its own layout on import; positions are always available via the tools regardless.

• OSINT provider transforms call third-party APIs — respect each provider's terms of service and rate limits. Without a key, those transforms are listed but skip with a clear "missing credential" message.

• Reports, layouts, scores, and recommendations are deterministic: the same graph + memory always yields the same output, so results are reproducible and shareable. (Timestamps in memory/events are the only non-deterministic field.)

• Investigation Memory is stored as a sidecar JSON member inside the .mtgx and is recovered on load — it survives save/load and merges, and is ignored by Maltego CE. Manually-added/CSV entities are "analyst-provided" (no discovering step) and scored with full source reliability.

• Real-time mode is optional: the event bus always buffers cheaply, and maltego_subscribe_events enables live callbacks. Over MCP stdio you retrieve events by polling maltego_get_recent_events (use since_seq for increments).

License

MIT