datalayer-mcp

MCP runtime library for the Datalayer platform. It turns curated GraphQL operation files into FastMCP tools backed by a GraphQL supergraph via datalayer-graphql.

Each .graphql file in your operations directory becomes one MCP tool. Custom tools can reuse the same supergraph client for computed or aggregated logic.

Features

• Config-driven bootstrap — mcp-config.yaml for supergraph URL, auth, and paths
• Operation discovery — scans a directory for *.graphql files at startup
• Auto tool registration — one MCP tool per named GraphQL operation
• Variable schemas — GraphQL variables mapped to Pydantic models → JSON Schema for MCP clients
• Custom tool support — @tool handlers (from fastmcp.tools) with shared get_client()
• .env loading — bearer token from a file next to your config (no manual export)
• Structured errors — federation-aware GraphQLError wrapped as OperationError
• Session-managed client — reuses HTTP connections via async with SupergraphClient

Requirements

• Python 3.13+
• uv (recommended)
• datalayer-graphql (workspace dependency)
• Network access to your GraphQL endpoint
• A valid bearer token accepted by that endpoint

Installation

From the monorepo root:

uv sync --package datalayer-mcp

With dev dependencies (for running tests):

uv sync --package datalayer-mcp --group dev

As a workspace dependency in another package:

dependencies = ["datalayer-mcp"]

[tool.uv.sources]
datalayer-mcp = { workspace = true }

Quick start

1. Project layout

my-mcp-server/
├── mcp-config.yaml
├── .env                      # SUPERGRAPH_TOKEN (gitignored)
├── server.py
├── operations/
│   └── Cves.graphql
└── tools/
    ├── __init__.py
    └── custom.py             # optional

A complete working example lives in examples/mcp/ at the monorepo root.

2. Configure supergraph connection

mcp-config.yaml

supergraph:
  url: https://graphql.example.com
  token_env: SUPERGRAPH_TOKEN
  client_name: my-mcp-server
  verify_ssl: true
  extra_headers: {}

operations:
  directory: operations

tools:
  dir: tools                  # optional directory of custom @tool files
  include_response_headers: false

.env (same directory as mcp-config.yaml):

SUPERGRAPH_TOKEN=eyJ...

Copy from the example:

cp examples/mcp/.env.example examples/mcp/.env

3. Add a GraphQL operation

operations/Cves.graphql

# List CVEs with pagination
query Cves($first: Int!) {
  cves(first: $first) {
    totalCount
    edges {
      node {
        title
        url
      }
    }
  }
}

Rules for operation files:

• Exactly one named operation per file (query, mutation, or subscription)
• The operation name in the GraphQL document becomes the MCP tool name (Cves)
• GraphQL docstrings ("""...""" or "...") or leading # comment lines become the tool description
• The filename does not have to match the operation name

4. Bootstrap the server

server.py

from pathlib import Path

from datalayer_mcp import create_server

CONFIG = Path(__file__).resolve().parent / "mcp-config.yaml"
mcp = create_server(config_path=CONFIG)

if __name__ == "__main__":
    mcp.run()

Use Path(__file__) so the config path works regardless of shell working directory.

5. Run

uv run --package datalayer-mcp examples/mcp/server.py

The server starts on stdio (standard MCP transport). After the FastMCP banner it waits for an MCP client — that idle state is normal.

---

Architecture

mcp-config.yaml + .env
        │
        ▼
  create_server()  ───►  DatalayerExtension(config_path)
                               │
                ┌──────────────┴──────────────┐
                ▼                             ▼
   FastMCP(lifespan=ext)               ext.mount(mcp)
        │                                     │
        ▼ (on server run)                     ├── auto-tools (LocalProvider)
   SupergraphClient                           ├── operation registry
   in lifespan_context                        └── FileSystemProvider (tools.dir)

Auto-generated vs custom tools

Type	Source	Example
Auto	Each operations/*.graphql file	Cves — runs the raw GraphQL operation
Custom	Python files in tools.dir	cve_count — calls Cves internally, returns a summary

Auto tool — registered from Cves.graphql:

# Internally: handler(variables: CvesVariables) → client.execute("Cves", ...)
# Returns: { "data": { "cves": { ... } } }

Custom tool — tools/custom.py:

from datalayer_mcp import get_client
from fastmcp.tools import tool


@tool
async def cve_count(first: int = 10) -> dict:
    """Return total CVE count from the Cves operation."""
    client = get_client()
    result = await client.execute(
        operation_name="Cves",
        variables={"first": first},
    )
    return {"total_count": result.data["cves"]["totalCount"]}

Custom tools live in the directory named by tools.dir. FastMCP's FileSystemProvider discovers @tool functions automatically; the files do not import the server.

---

Configuration reference

All paths in mcp-config.yaml are resolved relative to the config file's directory, not the shell cwd.

supergraph

Field	Type	Default	Description
url	URL	—	Supergraph endpoint
token_env	string	SUPERGRAPH_TOKEN	Env var name for bearer token
client_name	string	Package default	Apollo apollographql-client-name header
client_version	string	latest	Apollo client version header
timeout	float	30.0	HTTP timeout in seconds
verify_ssl	bool	true	TLS certificate verification
extra_headers	object	{}	Additional HTTP headers on every request

operations

Field	Type	Default	Description
directory	string	operations	Folder containing *.graphql files

tools

Field	Type	Default	Description
dir	string | null	null	Directory of custom @tool files, relative to the config file
include_response_headers	bool	false	Include router HTTP headers in auto-tool responses

When include_response_headers: true, auto-tools return:

{
  "data": { },
  "response_headers": { "content-type": "application/json" },
  "extensions": { }
}

Environment variables

Loaded from {config_dir}/.env, then {cwd}/.env as fallback. Existing shell variables are not overwritten.

Variable	Required	Description
SUPERGRAPH_TOKEN	Yes (for live calls)	Bearer token accepted by the configured endpoint

---

API reference

Public exports

from datalayer_mcp import (
    create_server,
    DatalayerExtension,
    execute_operation,
    get_client,
    SupergraphError,
    SupergraphConnectionError,
    SupergraphHTTPError,
    GraphQLError,
    GraphQLErrorDetail,
    OperationError,
)

create_server(name="datalayer_mcp", *, config_path, **fastmcp_kwargs)

Loads config, constructs a DatalayerExtension, passes it as lifespan to a new FastMCP instance (forwarding name and extra fastmcp_kwargs), mounts auto-tools and custom tools from tools.dir, and returns the FastMCP instance.

from pathlib import Path
from datalayer_mcp import create_server

mcp = create_server("my-mcp", config_path=Path("mcp-config.yaml").resolve())
mcp.run()

Startup validation:

• Operations directory must exist
• Each .graphql file must have valid syntax and exactly one named operation
• No duplicate operation names across files
• MCP parser and datalayer-graphql loader must agree on operation names
• Custom @tool names in tools.dir must not collide with operation names

get_client()

Returns the lifespan-managed SupergraphClient. Only available from within a tool handler while the MCP server is running — reads from FastMCP's lifespan_context via get_context(), with no module-level client global.

from datalayer_mcp import get_client

client = get_client()
result = await client.execute(operation_name="Cves", variables={"first": 10})

Raises RuntimeError if called outside an active MCP server session.

execute_operation(operation_name, variables=None)

Executes a registered GraphQL operation against the supergraph using the query string stored in the module-level operation registry. Must be called within an active tool handler (where get_client() is available).

from datalayer_mcp import execute_operation

result = await execute_operation("Cves", {"first": 10})

Raises ValueError if the operation name is not registered, or RuntimeError if called outside an active MCP server session.

DatalayerExtension(config_path)

A FastMCP Lifespan: opens a session-managed SupergraphClient on server startup and closes it on shutdown, storing it under lifespan_context["supergraph_client"]. create_server() wires this up for you; construct it directly for custom FastMCP setups:

from fastmcp import FastMCP
from datalayer_mcp import DatalayerExtension

ext = DatalayerExtension("mcp-config.yaml")
mcp = FastMCP("my-server", lifespan=ext)
ext.mount(mcp)

---

GraphQL variables → MCP input schema

Operation variables are parsed with graphql-core and converted to dynamic Pydantic models:

GraphQL type	Python / JSON Schema
String, ID	string
Int	integer
Float	number
Boolean	boolean
[T]	array
Input object	object (untyped in MVP)
Enum	string (no enum constraint in MVP)

Non-null (!) variables are required. Auto-tools accept a single variables object matching the Pydantic model (e.g. CvesVariables(first: int)).

Apollo Router validates input objects and enums at execution time even when MCP schemas are loose.

---

Error handling

GraphQL and HTTP errors from datalayer-graphql are wrapped in OperationError for auto-generated tools:

from datalayer_mcp import OperationError

try:
    ...
except OperationError as exc:
    print(exc.payload)
    # {
    #   "error_type": "GraphQLError",
    #   "message": "...",
    #   "errors": [{"message": "...", "code": "SUBREQUEST_HTTP_ERROR", "service": "..."}],
    #   "partial_data": ...
    # }

Exception	When
SupergraphConnectionError	Endpoint unavailable or network failure
SupergraphHTTPError	HTTP 401/403/503 (status_code attribute)
GraphQLError	GraphQL errors in response (may include partial federation data)
OperationError	Wrapper raised by auto-tools

In custom tools, catch GraphQLError directly or let errors propagate to the MCP client.

---

Connect to Cursor

Add to Cursor MCP settings (.cursor/mcp.json or Settings → MCP):

{
  "mcpServers": {
    "datalayer": {
      "command": "uv",
      "args": [
        "run",
        "--package",
        "datalayer-mcp",
        "examples/mcp/server.py"
      ],
      "cwd": "/absolute/path/to/datalayer-python-monorepo"
    }
  }
}

Put SUPERGRAPH_TOKEN in examples/mcp/.env. Restart Cursor after changing MCP config.

Available tools with the example project:

• Cves — auto-generated from poc/operations/Cves.graphql
• cve_count — custom tool from poc/tools/custom.py

---

Development

Package layout

packages/datalayer-mcp/
├── src/datalayer_mcp/
│   ├── __init__.py       # Public API
│   ├── config.py         # mcp-config.yaml → McpConfig
│   ├── env.py            # .env loading
│   ├── parse.py          # .graphql → ParsedOperation
│   ├── scan.py           # AST collision detection for tools.dir
│   ├── schema.py         # GraphQL variables → Pydantic models
│   ├── tools.py          # Auto-tool registration
│   ├── server.py         # create_server() orchestration
│   ├── extension.py      # DatalayerExtension (FastMCP Lifespan)
│   ├── runtime.py        # get_client() and execute_operation()
│   └── errors.py         # OperationError formatting
├── tests/                # Unit tests (no network)
└── pyproject.toml

Runnable POC server: `examples/mcp/` at the monorepo root.

Run tests

From monorepo root:

uv run --package datalayer-mcp pytest packages/datalayer-mcp/tests -v

From the package directory:

cd packages/datalayer-mcp
uv run pytest tests -v

Tests cover env loading, operation parsing, schema mapping, error formatting, duplicate detection, and mocked tool handlers. No network access or live supergraph is required.

Smoke-test server bootstrap

cp examples/mcp/.env.example examples/mcp/.env   # set SUPERGRAPH_TOKEN

uv run --package datalayer-mcp python -c "
from pathlib import Path
from datalayer_mcp import create_server
create_server(config_path=Path('examples/mcp/mcp-config.yaml').resolve())
print('create_server OK')
"

Run the example server

uv run --package datalayer-mcp examples/mcp/server.py

Live supergraph tests

Live integration tests live in datalayer-graphql:

cd packages/datalayer-graphql
cp .env.example .env
uv run pytest tests/test_smoke_supergraph.py -v -s

Adding a new auto-tool

1. Create operations/MyOperation.graphql with one named operation
2. Restart the MCP server
3. The tool appears automatically as MyOperation

Adding a custom tool

1. Create a .py file in the directory named by tools.dir in mcp-config.yaml
2. Use @tool (from fastmcp.tools) and get_client() as shown above
3. Restart the MCP server; the tool is discovered automatically

---

Obtaining a supergraph token

Fetch a client-credentials token from your OAuth 2.0 provider:

curl -X POST \
  "https://auth.example.com/oauth2/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" \
  -d "client_id=YOUR_CLIENT_ID" \
  -d "client_secret=YOUR_SECRET" \
  -d "scope=my-api-scope"

Copy access_token into .env as SUPERGRAPH_TOKEN.

---

Relationship to datalayer-graphql

Concern	Package
HTTP client, auth, Apollo headers	datalayer-graphql
Operation file loading (registry)	datalayer-graphql.load_operations()
MCP tool registration, config, .env	datalayer-mcp
Variable → JSON Schema for MCP	datalayer-mcp
Custom tool integration	datalayer-mcp

datalayer-mcp does not implement GraphQL HTTP directly — all supergraph communication goes through SupergraphClient.

---

Troubleshooting

Problem	Likely cause	Fix
FileNotFoundError: mcp-config.yaml	Wrong cwd	Use Path(__file__).parent / "mcp-config.yaml"
Missing env var: SUPERGRAPH_TOKEN	No .env or expired token	Copy .env.example, refresh token
FileNotFoundError: tools.dir not found	tools.dir path missing	Create the directory or drop the tools.dir key
ValueError: Custom tool name collides	@tool name matches an operation	Rename the custom tool or the operation
ValidationError: instructions	FastMCP(name, lifespan) as positional arg	Use FastMCP(name, lifespan=lifespan)
Operation registry mismatch	Parser disagreement	Check file encoding; both loaders strip whitespace
Server idle after banner	Normal stdio behavior	Connect via Cursor MCP client
Tool call fails	Endpoint unavailable or bad token	Check endpoint access and refresh SUPERGRAPH_TOKEN

---

License

See LICENSE.txt.