feat: add general GraphQL tools to MCP server
Add two new MCP tools for flexible GraphQL operations: - get_graphql_schema: Expose complete schema via introspection - execute_graphql_query: Execute arbitrary queries and mutations These tools complement existing dedicated member tools by enabling AI agents to: - Discover the GraphQL schema dynamically - Construct custom queries with specific field selection - Handle complex queries without requiring new dedicated tools - Work with query variables for parameterized operations The implementation reuses the existing GraphQLClient class and adds proper error handling for API unavailability and validation errors. Schema introspection is formatted as human-readable text while query results are returned as formatted JSON. Updated README.md with: - Documentation for both new tools - Usage examples for schema discovery and query execution - Guidance on when to use general vs dedicated tools 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -5,6 +5,7 @@ via the Model Context Protocol (MCP), connecting to the Clubber GraphQL API.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
@@ -13,6 +14,57 @@ from mcp.server import Server
|
||||
from mcp.server.stdio import stdio_server
|
||||
from mcp.types import TextContent, Tool
|
||||
|
||||
# GraphQL introspection query to fetch the complete schema
|
||||
INTROSPECTION_QUERY = """
|
||||
query IntrospectionQuery {
|
||||
__schema {
|
||||
queryType { name }
|
||||
mutationType { name }
|
||||
types {
|
||||
kind
|
||||
name
|
||||
description
|
||||
fields(includeDeprecated: true) {
|
||||
name
|
||||
description
|
||||
type {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
}
|
||||
}
|
||||
args {
|
||||
name
|
||||
description
|
||||
type {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
inputFields {
|
||||
name
|
||||
description
|
||||
type {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
class GraphQLClient:
|
||||
"""HTTP client for communicating with the Clubber GraphQL API."""
|
||||
@@ -181,6 +233,33 @@ class MemberManagementServer:
|
||||
"required": ["id"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="get_graphql_schema",
|
||||
description="Get the complete GraphQL schema definition via introspection. This allows AI agents to discover all available types, queries, mutations, and their fields dynamically.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"required": [],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="execute_graphql_query",
|
||||
description="Execute an arbitrary GraphQL query or mutation. Use this for complex queries beyond the dedicated member tools. The query can include variables for parameterization.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "The GraphQL query or mutation string",
|
||||
},
|
||||
"variables": {
|
||||
"type": "object",
|
||||
"description": "Optional variables for the query as a JSON object",
|
||||
},
|
||||
},
|
||||
"required": ["query"],
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
@self.server.call_tool()
|
||||
@@ -195,6 +274,12 @@ class MemberManagementServer:
|
||||
return await self._create_member(arguments)
|
||||
elif name == "update_member":
|
||||
return await self._update_member(arguments)
|
||||
elif name == "get_graphql_schema":
|
||||
return await self._get_graphql_schema()
|
||||
elif name == "execute_graphql_query":
|
||||
return await self._execute_graphql_query(
|
||||
arguments["query"], arguments.get("variables")
|
||||
)
|
||||
else:
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
except Exception as e:
|
||||
@@ -330,6 +415,144 @@ class MemberManagementServer:
|
||||
)
|
||||
]
|
||||
|
||||
async def _get_graphql_schema(self) -> List[TextContent]:
|
||||
"""Get the complete GraphQL schema via introspection."""
|
||||
try:
|
||||
data = await self.graphql.query(INTROSPECTION_QUERY)
|
||||
schema = data.get("__schema", {})
|
||||
|
||||
# Format schema as readable text
|
||||
output = ["GraphQL Schema\n" + "=" * 50 + "\n"]
|
||||
|
||||
# Query type
|
||||
if schema.get("queryType"):
|
||||
output.append(f"Query Type: {schema['queryType']['name']}\n")
|
||||
|
||||
# Mutation type
|
||||
if schema.get("mutationType"):
|
||||
output.append(f"Mutation Type: {schema['mutationType']['name']}\n")
|
||||
|
||||
output.append("\nAvailable Types:\n" + "-" * 50)
|
||||
|
||||
# List all types (excluding internal types)
|
||||
for type_info in schema.get("types", []):
|
||||
type_name = type_info.get("name", "")
|
||||
# Skip internal GraphQL types
|
||||
if type_name.startswith("__"):
|
||||
continue
|
||||
|
||||
output.append(f"\n{type_info.get('kind', 'OBJECT')}: {type_name}")
|
||||
|
||||
if type_info.get("description"):
|
||||
output.append(f" Description: {type_info['description']}")
|
||||
|
||||
# List fields
|
||||
fields = type_info.get("fields", [])
|
||||
if fields:
|
||||
output.append(" Fields:")
|
||||
for field in fields:
|
||||
field_type = self._format_type(field.get("type", {}))
|
||||
field_desc = (
|
||||
f" - {field.get('description')}"
|
||||
if field.get("description")
|
||||
else ""
|
||||
)
|
||||
output.append(
|
||||
f" - {field['name']}: {field_type}{field_desc}"
|
||||
)
|
||||
|
||||
# List field arguments
|
||||
args = field.get("args", [])
|
||||
if args:
|
||||
output.append(" Arguments:")
|
||||
for arg in args:
|
||||
arg_type = self._format_type(arg.get("type", {}))
|
||||
arg_desc = (
|
||||
f" - {arg.get('description')}"
|
||||
if arg.get("description")
|
||||
else ""
|
||||
)
|
||||
output.append(
|
||||
f" - {arg['name']}: {arg_type}{arg_desc}"
|
||||
)
|
||||
|
||||
# List input fields
|
||||
input_fields = type_info.get("inputFields", [])
|
||||
if input_fields:
|
||||
output.append(" Input Fields:")
|
||||
for field in input_fields:
|
||||
field_type = self._format_type(field.get("type", {}))
|
||||
field_desc = (
|
||||
f" - {field.get('description')}"
|
||||
if field.get("description")
|
||||
else ""
|
||||
)
|
||||
output.append(
|
||||
f" - {field['name']}: {field_type}{field_desc}"
|
||||
)
|
||||
|
||||
return [TextContent(type="text", text="\n".join(output))]
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
if "Failed to connect" in error_msg:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Error: Could not connect to GraphQL API.\n\n"
|
||||
f"Please ensure the API is running at {self.graphql.api_url}\n"
|
||||
f"Start it with: uv run uvicorn src.main:app --host 127.0.0.1 --port 8000",
|
||||
)
|
||||
]
|
||||
raise
|
||||
|
||||
async def _execute_graphql_query(
|
||||
self, query: str, variables: Optional[Dict[str, Any]] = None
|
||||
) -> List[TextContent]:
|
||||
"""Execute an arbitrary GraphQL query or mutation."""
|
||||
try:
|
||||
data = await self.graphql.query(query, variables)
|
||||
# Format the result as pretty JSON
|
||||
result_json = json.dumps(data, indent=2)
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Query executed successfully:\n\n{result_json}",
|
||||
)
|
||||
]
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
if "GraphQL errors:" in error_msg:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"GraphQL validation error:\n\n{error_msg}\n\n"
|
||||
f"Please check your query syntax and field names.",
|
||||
)
|
||||
]
|
||||
elif "Failed to connect" in error_msg:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Error: Could not connect to GraphQL API.\n\n"
|
||||
f"Please ensure the API is running at {self.graphql.api_url}",
|
||||
)
|
||||
]
|
||||
raise
|
||||
|
||||
def _format_type(self, type_info: Dict[str, Any]) -> str:
|
||||
"""Format a GraphQL type for display."""
|
||||
kind = type_info.get("kind", "")
|
||||
name = type_info.get("name", "")
|
||||
|
||||
if kind == "NON_NULL":
|
||||
inner = self._format_type(type_info.get("ofType", {}))
|
||||
return f"{inner}!"
|
||||
elif kind == "LIST":
|
||||
inner = self._format_type(type_info.get("ofType", {}))
|
||||
return f"[{inner}]"
|
||||
else:
|
||||
return name or "Unknown"
|
||||
|
||||
def _format_member(self, member: Dict[str, Any]) -> str:
|
||||
"""Format a single member for display."""
|
||||
lines = [
|
||||
|
||||
Reference in New Issue
Block a user