# Directory Structure
```
├── age-mcp-server.rb
├── docs
│ └── claude_desktop_config_sample.json
├── images
│ ├── .DS_Store
│ ├── Claude_Win.png
│ ├── query_01.png
│ ├── query_02.png
│ ├── query_03.png
│ ├── query_04.png
│ ├── query_05.png
│ ├── query_06.png
│ ├── query_07.png
│ ├── query_08.png
│ ├── query_09.png
│ ├── vscode_chat_01.png
│ ├── vscode_chat_02.png
│ └── vscode_mcp_settings.png
├── LICENSE
├── pyproject.toml
├── README.md
├── src
│ └── age_mcp_server
│ ├── __init__.py
│ └── server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# AGE-MCP-Server
Apache AGE MCP Server
[Apache AGE™](https://age.apache.org/) is a PostgreSQL Graph database compatible with PostgreSQL's distributed assets and leverages graph data structures to analyze and use relationships and patterns in data.
[Azure Database for PostgreSQL](https://azure.microsoft.com/en-us/services/postgresql/) is a managed database service that is based on the open-source Postgres database engine.
[Introducing support for Graph data in Azure Database for PostgreSQL (Preview)](https://techcommunity.microsoft.com/blog/adforpostgresql/introducing-support-for-graph-data-in-azure-database-for-postgresql-preview/4275628).
## Table of Contents
- [Prerequisites](#prerequisites)
- [Install](#install)
- [Usage with Claude](#usage-with-claude)
- [Usage with Visual Studio Code Insiders](#usage-with-visual-studio-code-insiders)
- [Write Operations](#write-operations)
- [Release Notes](#release-notes)
- [For More Information](#for-more-information)
- [License](#license)
## Prerequisites
- Python 3.13 and above
- This module runs on [psycopg](https://www.psycopg.org/)
- Enable the Apache AGE extension in your Azure Database for PostgreSQL instance. Login Azure Portal, go to 'server parameters' blade, and check 'AGE" on within 'azure.extensions' and 'shared_preload_libraries' parameters. See, above blog post for more information.
- Load the AGE extension in your PostgreSQL database.
```sql
CREATE EXTENSION IF NOT EXISTS age CASCADE;
```
- Claude
Download from [Claude Desktop Client](https://claude.ai/download) or,
```bash
brew install claude
```
- Visual Studio Code Insiders
Download from [Visual Studio Code](https://code.visualstudio.com/download) or,
```bash
brew intall visual-studio-code
```
## Install
- with brew
```bash
brew tap rioriost/age-mcp-server
brew install age-mcp-server
```
- with uv
```bash
uv init your_project
cd your_project
uv venv
source .venv/bin/activate
uv add age-mcp-server
```
- with python venv on macOS / Linux
```bash
mkdir your_project
cd your_project
python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install age-mcp-server
```
- with python venv on Windows
```bash
mkdir your_project
cd your_project
python -m venv venv
.\venv\Scripts\activate
python -m pip install age-mcp-server
```
## Usage with Claude
- on macOS
`claude_desktop_config.json` is located in `~/Library/Application Support/Claude/`.
- on Windows
You need to create a new `claude_desktop_config.json` under `%APPDATA%\Claude`.
- Homebrew on macOS
Homebrew installs `age-mcp-server` into $PATH.
```json
{
"mcpServers": {
"age-manager": {
"command": "age-mcp-server",
"args": [
"--pg-con-str",
"host=your_server.postgres.database.azure.com port=5432 dbname=postgres user=your_username password=your_password",
]
}
}
}
```
- uv / Pyhon venv
On macOS:
```json
{
"mcpServers": {
"age-manager": {
"command": "/Users/your_username/.local/bin/uv",
"args": [
"--directory",
"/path/to/your_project",
"run",
"age-mcp-server",
"--pg-con-str",
"host=your_server.postgres.database.azure.com port=5432 dbname=postgres user=your_username password=your_password",
]
}
}
}
```
On Windows:
```json
{
"mcpServers": {
"age-manager": {
"command": "C:\\Users\\USER\\.local\\bin\\uv.exe",
"args": [
"--directory",
"C:\\path\\to\\your_project",
"run",
"age-mcp-server",
"--pg-con-str",
"host=your_server.postgres.database.azure.com port=5432 dbname=postgres user=your_username password=your_password",
]
}
}
}
```
If you need to hide the password or to use Entra ID, you can set `--pg-con-str` as follows.
```
{
"mcpServers": {
"age-manager": {
...
"--pg-con-str",
"host=your_server.postgres.database.azure.com port=5432 dbname=postgres user=your_username",
...
]
}
}
}
```
And, you need to set `PGPASSWORD` env variable, or to [install Azure CLI](https://learn.microsoft.com/en-us/cli/azure/install-azure-cli) and [sign into Azure](https://learn.microsoft.com/en-us/cli/azure/authenticate-azure-cli) with your Azure account.
After saving `claude_desktop_config.json`, start Claude Desktop Client.
## Usage with Visual Studio Code
After installing, [Preferences]->[Settings] and input `mcp` to [Search settings].
Edit the settings.json as followings:
```json
{
"mcp": {
"inputs": [],
"servers": {
"age-manager": {
"command": "/Users/your_user_name/.local/bin/uv",
"args": [
"--directory",
"/path/to/your_project",
"run",
"age-mcp-server",
"--pg-con-str",
"host=your_server.postgres.database.azure.com port=5432 dbname=postgres user=your_username password=your_password",
"--debug"
]
}
}
}
}
```
And then, you'll see `start` to start the AGE MCP Server.
Switch the Chat window to `agent` mode.
Now, you can play with your graph data via Visual Studio Code!
## Write Operations
AGE-MCP-Server prohibits write operations by default for safety. If you want to enable write operations, you can use the `--allow-write` flag.
```json
{
"mcpServers": {
"age-manager": {
"command": "age-mcp-server",
"args": [
"--pg-con-str",
"host=your_server.postgres.database.azure.com port=5432 dbname=postgres user=your_username password=your_password",
"--allow-write"
]
}
}
}
```
## Release Notes
### 0.2.32 Release
- Dependency Update
### 0.2.31 Release
- Dependency Update
### 0.2.30 Release
- Dependency Update
### 0.2.29 Release
- Dependency Update
### 0.2.28 Release
- Dependency Update
### 0.2.27 Release
- Dependency Update
### 0.2.26 Release
- Dependency Update
### 0.2.25 Release
- Dependency Update
### 0.2.24 Release
- Dependency Update
### 0.2.23 Release
- Dependency Update
### 0.2.22 Release
- Dependency Update
### 0.2.21 Release
- Dependency Update
### 0.2.20 Release
- Dependency Update
### 0.2.19 Release
- Dependency Update
### 0.2.18 Release
- Dependency Update
### 0.2.17 Release
- Dependency Update
### 0.2.16 Release
- Dependency Update
### 0.2.15 Release
- Dependency Update
### 0.2.14 Release
- Dependency Update
### 0.2.13 Release
- Dependency Update
### 0.2.12 Release
- Dependency Update
### 0.2.11 Release
- Dependency Update
### 0.2.10 Release
- Dependency Update
### 0.2.9 Release
- Dependency Update
### 0.2.8 Release
- Add support for VSCode(Stable)
### 0.2.7 Release
- Add support for VSCode Insiders
### 0.2.6 Release
- Fix a typo
### 0.2.5 Release
- Support connection with Entra ID
### 0.2.4 Release
- Dependency Update
### 0.2.3 Release
- Dependency Update
### 0.2.2 Release
- Drop a conditional test of `CREATE` operation by adding `RETURN` to the description for `write-age-cypher` tool.
### 0.2.1 Release
- Fix a bug in node/edge creation
### 0.2.0 Release
- Add multiple graph support
- Add graph creation and deletion support
- Obsolete `--graph-name` argument
### 0.1.8 Release
- Add `--allow-write` flag
### 0.1.7 Release
- Add Windows support
### 0.1.6 Release
- Fix parser for `RETURN` values
### 0.1.5 Release
- Draft release
### 0.1.4 Release
- Draft release
### 0.1.3 Release
- Draft release
### 0.1.2 Release
- Draft release
### 0.1.1 Release
- Draft release
### 0.1.0a1 Release
- Draft release
## For More Information
- Apache AGE : https://age.apache.org/
- GitHub : https://github.com/apache/age
- Document : https://age.apache.org/age-manual/master/index.html
## License
MIT License
```
--------------------------------------------------------------------------------
/docs/claude_desktop_config_sample.json:
--------------------------------------------------------------------------------
```json
{
"mcpServers": {
"age-manager": {
"command": "/Users/your_username/.local/bin/uv",
"args": [
"--directory",
"/path/to/your_project",
"run",
"age-mcp-server",
"--pg-con-str",
"host=your_server.postgres.database.azure.com port=5432 dbname=postgres user=your_username password=your_password"
]
}
}
}
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "age-mcp-server"
authors = [{ name = "Rio Fujita", email = "[email protected]" }]
version = "0.2.32"
license = { file = "LICENSE" }
description = "Apache AGE MCP Server"
readme = { file = "README.md", content-type = "text/markdown" }
requires-python = ">=3.13"
dependencies = [
"agefreighter>=1.0.20",
"mcp>=1.21.1",
"ply>=3.11",
"psycopg[binary,pool]>=3.2.11",
]
[project.urls]
Homepage = "https://github.com/rioriost/homebrew-age-mcp-server"
Issues = "https://github.com/rioriost/homebrew-age-mcp-server/issues"
[project.scripts]
age-mcp-server = "age_mcp_server:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[dependency-groups]
dev = []
```
--------------------------------------------------------------------------------
/age-mcp-server.rb:
--------------------------------------------------------------------------------
```ruby
class AgeMcpServer < Formula
include Language::Python::Virtualenv
desc "Apache AGE MCP Server"
homepage "https://github.com/rioriost/homebrew-age-mcp-server/"
url "https://files.pythonhosted.org/packages/64/d4/20d05d55ae1df2e7fdd967cbb5ffffb637dfd25d15d8ac27a1470a67aa6d/age_mcp_server-0.2.32.tar.gz"
sha256 "e1358556f5fc3edeb0d295107eaf2916c21da0b355bb355c2c13fd7b76c29ae8"
license "MIT"
depends_on "[email protected]"
resource "agefreighter" do
url "https://files.pythonhosted.org/packages/df/3d/5f701f1135cef1ced759c319d6f5c0454c379e5e427d4067e20e03017be9/agefreighter-1.0.20.tar.gz"
sha256 "1a7994b684fe3e91727e632c1883751646bb157b02ff3047f6f73e68912cafd5"
end
resource "ply" do
url "https://files.pythonhosted.org/packages/e5/69/882ee5c9d017149285cab114ebeab373308ef0f874fcdac9beb90e0ac4da/ply-3.11.tar.gz"
sha256 "00c7c1aaa88358b9c765b6d3000c6eec0ba42abca5351b095321aef446081da3"
end
def install
virtualenv_install_with_resources
system libexec/"bin/python", "-m", "pip", "install", "psycopg[binary,pool]", "mcp"
end
test do
system "#{bin}/age-mcp-server", "--help"
end
end
```
--------------------------------------------------------------------------------
/src/age_mcp_server/__init__.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import asyncio
import logging
import os
import subprocess
import sys
from . import server
logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
def main() -> None:
"""Main entry point as command line tool."""
parser = argparse.ArgumentParser(description="Apache AGE MCP Server")
parser.add_argument(
"--pg-con-str",
type=str,
default=os.environ.get("PG_CONNECTION_STRING", ""),
help="Connection string of the Azure Database for PostgreSQL",
)
parser.add_argument(
"-w",
"--allow-write",
action="store_true",
default=False,
help="Allow write operations",
)
parser.add_argument(
"--debug", action="store_true", default=False, help="Enable debug logging"
)
args = parser.parse_args()
if not args.pg_con_str:
print("Error: PostgreSQL connection string is required.")
sys.exit(1)
conn_dict = dict(item.split("=", 1) for item in args.pg_con_str.split())
if not conn_dict.get("password"):
# Try to get password from env variable
conn_dict["password"] = os.environ.get("PGPASSWORD", "")
if not conn_dict["password"]:
# Try to get password using azure cli
conn_dict["password"] = subprocess.check_output(
[
"az",
"account",
"get-access-token",
"--resource",
"https://ossrdbms-aad.database.windows.net",
"--query",
"accessToken",
"--output",
"tsv",
],
stderr=subprocess.DEVNULL,
text=True,
).strip()
if not conn_dict["password"]:
print(
"Error: Could not find PGPASSWORD env var or Entra ID token to connect the server."
)
sys.exit(1)
asyncio.run(
server.main(
pg_con_str=" ".join({f"{k}={v}" for k, v in conn_dict.items()}),
allow_write=args.allow_write,
log_level=logging.DEBUG if args.debug else logging.INFO,
)
)
__all__ = ["main", "server"]
```
--------------------------------------------------------------------------------
/src/age_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
import logging
import re
import sys
from typing import Any
from mcp.server import NotificationOptions, Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
from psycopg import Connection
from psycopg.rows import dict_row
from agefreighter.cypherparser import CypherParser
logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
class CypherQueryFormatter:
"""Utility class for formatting Cypher queries for Apache AGE."""
@staticmethod
def format_query(graph_name: str, cypher_query: str, allow_write: bool) -> str:
"""
Format the provided Cypher query for Apache AGE.
Raises:
ValueError: If the query is unsafe or incorrectly formatted.
"""
if not allow_write:
if not CypherQueryFormatter.is_safe_cypher_query(cypher_query):
raise ValueError("Unsafe query")
# Append LIMIT 50 if no limit is specified.
if "limit" not in cypher_query.lower():
cypher_query += " LIMIT 50"
# Claude misunderstands the Cypher definition
if "cast" in cypher_query.lower():
raise ValueError("'CAST' is not a reserved keyword in Cypher")
returns = CypherQueryFormatter.get_return_values(cypher_query)
log.debug(f"Return values: {returns}")
# Check for parameterized query usage.
if re.findall(r"\$(\w+)", cypher_query):
raise ValueError("Parameterized query")
if returns:
ag_types = ", ".join([f"{r} agtype" for r in returns])
return f"SELECT * FROM cypher('{graph_name}', $$ {cypher_query} $$) AS ({ag_types});"
else:
raise ValueError("No return values specified")
@staticmethod
def is_safe_cypher_query(cypher_query: str) -> bool:
"""
Ensure the Cypher query does not contain dangerous commands.
Returns:
bool: True if safe, False otherwise.
"""
tokens = cypher_query.split()
unsafe_keywords = ["add", "create", "delete", "merge", "remove", "set"]
return all(token.lower() not in unsafe_keywords for token in tokens)
@staticmethod
def get_return_values(cypher_query: str) -> list:
parser = CypherParser()
try:
result = parser.parse(cypher_query)
except Exception as e:
log.error(f"Failed to parse Cypher query: {e}")
return []
for op, opr, *_ in result:
log.debug(f"Returning values from query: {opr}")
if op == "RETURN" or op == "RETURN_DISTINCT":
results = []
for v in opr:
if isinstance(v, str):
results.append(v.split(".")[0])
elif isinstance(v, tuple):
match v[0]:
case "alias":
results.append(v[-1])
case "property":
results.append(v[-1])
case "func_call":
results.append(v[1])
case "":
pass
return list(set(results))
return []
class PostgreSQLAGE:
def __init__(self, pg_con_str: str, allow_write: bool, log_level: int):
"""Initialize connection to the PostgreSQL database"""
log.setLevel(log_level)
log.debug(f"Initializing database connection to {pg_con_str}")
self.pg_con_str = pg_con_str
self.allow_write = allow_write
self.con: Connection
try:
self.con = Connection.connect(
self.pg_con_str
+ " options='-c search_path=ag_catalog,\"$user\",public'"
)
except Exception as e:
log.error(f"Failed to connect to PostgreSQL database: {e}")
sys.exit(1)
def _execute_query(
self, graph_name: str, query: str, params: dict[str, Any] | None = None
) -> list[dict[str, Any]]:
"""Execute a Cypher query and return results as a list of dictionaries"""
log.debug(f"Executing query: {query}")
try:
cur = self.con.cursor(row_factory=dict_row)
cypher_query = CypherQueryFormatter.format_query(
graph_name=graph_name,
cypher_query=query,
allow_write=self.allow_write,
)
log.debug(f"Formatted query: {cypher_query}")
cur.execute(cypher_query, params)
results = cur.fetchall()
cur.execute("COMMIT")
count = len(results)
if CypherQueryFormatter.is_safe_cypher_query(query):
log.debug(f"Read query returned {count} rows")
return results
else:
log.debug(f"Write query affected {count}")
return [count]
except Exception as e:
log.error(f"Database error executing query: {e}\n{query}")
self.con.rollback() # Roll back to clear the error state
raise
def _execute_sql(self, query: str) -> list[dict[str, Any]]:
"""Execute a standard query and return results as a list of dictionaries"""
log.debug(f"Executing query: {query}")
try:
cur = self.con.cursor(row_factory=dict_row)
cur.execute(query)
results = cur.fetchall()
cur.execute("COMMIT")
return results
except Exception as e:
log.error(f"Database error executing query: {e}\n{query}")
self.con.rollback() # Roll back to clear the error state
raise
async def main(pg_con_str: str, allow_write: bool, log_level: int) -> None:
log.setLevel(log_level)
log.info(f"Connecting to PostgreSQL with connection string: {pg_con_str}")
db = PostgreSQLAGE(
pg_con_str=pg_con_str,
allow_write=allow_write,
log_level=log_level,
)
server = Server("age-manager")
# Register handlers
log.debug("Registering handlers")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available tools"""
return [
types.Tool(
name="read-age-cypher",
description="Execute a Cypher query on the AGE",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Cypher read query to execute",
},
"graph_name": {
"type": "string",
"description": "Name of the graph to operate",
},
},
"required": ["query", "graph_name"],
},
),
types.Tool(
name="write-age-cypher",
description="Execute a write Cypher query on the AGE",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Cypher write query to execute, including 'RETURN' statement",
},
"graph_name": {
"type": "string",
"description": "Name of the graph to operate",
},
},
"required": ["query", "graph_name"],
},
),
types.Tool(
name="create-age-graph",
description="Create a new graph in the AGE",
inputSchema={
"type": "object",
"properties": {
"graph_name": {
"type": "string",
"description": "Name of the graph to create",
},
},
"required": ["graph_name"],
},
),
types.Tool(
name="drop-age-graph",
description="Drop a graph in the AGE",
inputSchema={
"type": "object",
"properties": {
"graph_name": {
"type": "string",
"description": "Name of the graph to drop",
},
},
"required": ["graph_name"],
},
),
types.Tool(
name="list-age-graphs",
description="List all graphs in the AGE",
inputSchema={
"type": "object",
},
),
types.Tool(
name="get-age-schema",
description="List all node types, their attributes and their relationships TO other node-types in the AGE",
inputSchema={
"type": "object",
"properties": {
"graph_name": {
"type": "string",
"description": "Name of the graph to create",
},
},
"required": ["graph_name"],
},
),
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""Handle tool execution requests"""
try:
if name == "get-age-schema":
node_results = db._execute_query(
graph_name=arguments["graph_name"],
query="""
MATCH (n)
UNWIND labels(n) AS label
RETURN DISTINCT label, collect(DISTINCT keys(n)) AS properties
""",
)
log.debug(f"Node results: {node_results}")
edge_results = db._execute_query(
graph_name=arguments["graph_name"],
query="""
MATCH (a)-[r]->(b)
RETURN DISTINCT type(r) AS rel_type, collect(DISTINCT labels(a)) AS from_labels, collect(DISTINCT labels(b)) AS to_labels
""",
)
log.debug(f"Edge results: {edge_results}")
nodes_dict = {}
for node in node_results:
label = node["label"].strip('"')
props = json.loads(node["properties"])
properties = (
props[0]
if props and isinstance(props, list) and len(props) > 0
else []
)
nodes_dict[label] = {
"label": label,
"properties": properties,
"relationships": {},
}
edges = []
for edge in edge_results:
rel_type = edge["rel_type"].strip('"')
from_labels = json.loads(edge["from_labels"])
to_labels = json.loads(edge["to_labels"])
from_labels = (
from_labels[0]
if from_labels and isinstance(from_labels, list)
else []
)
to_labels = (
to_labels[0]
if to_labels and isinstance(to_labels, list)
else []
)
edges.append(
{
"rel_type": rel_type,
"from_labels": from_labels,
"to_labels": to_labels,
}
)
for from_label in from_labels:
if from_label in nodes_dict and to_labels:
nodes_dict[from_label]["relationships"][rel_type] = (
to_labels[0]
)
for to_label in to_labels:
if to_label in nodes_dict and from_labels:
nodes_dict[to_label]["relationships"][rel_type] = (
from_labels[0]
)
nodes = list(nodes_dict.values())
return [
types.TextContent(
type="text", text=str({"nodes": nodes, "edges": edges})
)
]
elif name == "create-age-graph":
if not allow_write:
raise PermissionError("Not allowed to create graph")
query = "SELECT create_graph('{}')".format(arguments["graph_name"])
log.info(f"Creating graph with name {arguments['graph_name']}")
results = db._execute_sql(query=query)
return [types.TextContent(type="text", text=str(results))]
elif name == "drop-age-graph":
if not allow_write:
raise PermissionError("Not allowed to drop graph")
query = "SELECT drop_graph('{}', True)".format(arguments["graph_name"])
log.info(f"Dropping graph with name {arguments['graph_name']}")
results = db._execute_sql(query=query)
return [types.TextContent(type="text", text=str(results))]
elif name == "list-age-graphs":
query = "SELECT name FROM ag_graph"
log.info("Listing graphs")
results = db._execute_sql(query=query)
return [types.TextContent(type="text", text=str(results))]
elif name == "read-age-cypher":
if not CypherQueryFormatter.is_safe_cypher_query(arguments["query"]):
raise ValueError("Only MATCH queries are allowed for read-query")
results = db._execute_query(
graph_name=arguments["graph_name"], query=arguments["query"]
)
return [types.TextContent(type="text", text=str(results))]
elif name == "write-age-cypher":
if CypherQueryFormatter.is_safe_cypher_query(arguments["query"]):
raise ValueError("Only write queries are allowed for write-query")
results = db._execute_query(
graph_name=arguments["graph_name"], query=arguments["query"]
)
return [types.TextContent(type="text", text=str(results))]
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
return [types.TextContent(type="text", text=f"Error: {str(e)}")]
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
log.info("Server running with stdio transport")
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="age",
server_version="0.2.8",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
```