feat: add MCP server for AI-powered member management
This change implements a Model Context Protocol (MCP) server that enables AI assistants to manage society members through natural language interactions. Changes: - Add MCP server implementation (src/mcp_server.py) - Implement 4 core tools: * list_members - Query all members * get_member - Get specific member by ID * create_member - Create new member * update_member - Update member information - Add mcp and httpx dependencies to pyproject.toml - Update README with comprehensive MCP server documentation - Create OpenSpec proposal in openspec/changes/add-mcp-server/ The MCP server connects to the GraphQL API via HTTP and provides structured tools that abstract away GraphQL complexity. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -0,0 +1,399 @@
|
||||
"""MCP Server for Clubber Member Management.
|
||||
|
||||
This server provides AI assistants with tools to manage society members
|
||||
via the Model Context Protocol (MCP), connecting to the Clubber GraphQL API.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import httpx
|
||||
from mcp.server import Server
|
||||
from mcp.server.stdio import stdio_server
|
||||
from mcp.types import TextContent, Tool
|
||||
|
||||
|
||||
class GraphQLClient:
|
||||
"""HTTP client for communicating with the Clubber GraphQL API."""
|
||||
|
||||
def __init__(self, api_url: str):
|
||||
"""Initialize the GraphQL client.
|
||||
|
||||
Args:
|
||||
api_url: The GraphQL API endpoint URL
|
||||
"""
|
||||
self.api_url = api_url
|
||||
self.client = httpx.AsyncClient(timeout=30.0)
|
||||
|
||||
async def query(
|
||||
self, query: str, variables: Optional[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Execute a GraphQL query.
|
||||
|
||||
Args:
|
||||
query: The GraphQL query string
|
||||
variables: Optional query variables
|
||||
|
||||
Returns:
|
||||
The GraphQL response data
|
||||
|
||||
Raises:
|
||||
Exception: If the query fails or returns errors
|
||||
"""
|
||||
payload = {"query": query}
|
||||
if variables:
|
||||
payload["variables"] = variables
|
||||
|
||||
try:
|
||||
response = await self.client.post(self.api_url, json=payload)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
if "errors" in result:
|
||||
error_messages = [e.get("message", str(e)) for e in result["errors"]]
|
||||
raise Exception(f"GraphQL errors: {', '.join(error_messages)}")
|
||||
|
||||
return result.get("data", {})
|
||||
except httpx.RequestError as e:
|
||||
raise Exception(f"Failed to connect to GraphQL API at {self.api_url}: {e}")
|
||||
|
||||
async def close(self):
|
||||
"""Close the HTTP client."""
|
||||
await self.client.aclose()
|
||||
|
||||
|
||||
class MemberManagementServer:
|
||||
"""MCP server providing member management tools."""
|
||||
|
||||
def __init__(self, api_url: str):
|
||||
"""Initialize the MCP server.
|
||||
|
||||
Args:
|
||||
api_url: The GraphQL API endpoint URL
|
||||
"""
|
||||
self.graphql = GraphQLClient(api_url)
|
||||
self.server = Server("clubber-mcp-server")
|
||||
self._register_handlers()
|
||||
|
||||
def _register_handlers(self):
|
||||
"""Register MCP protocol handlers."""
|
||||
|
||||
@self.server.list_tools()
|
||||
async def list_tools() -> List[Tool]:
|
||||
"""List available tools."""
|
||||
return [
|
||||
Tool(
|
||||
name="list_members",
|
||||
description="List all members in the society with their complete information",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"required": [],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="get_member",
|
||||
description="Get detailed information about a specific member by their ID",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {
|
||||
"type": "integer",
|
||||
"description": "The member's unique ID",
|
||||
}
|
||||
},
|
||||
"required": ["id"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="create_member",
|
||||
description="Create a new member. Only firstName is required, all other fields are optional.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"firstName": {
|
||||
"type": "string",
|
||||
"description": "Member's first name (required)",
|
||||
},
|
||||
"lastName": {
|
||||
"type": "string",
|
||||
"description": "Member's last name",
|
||||
},
|
||||
"street": {
|
||||
"type": "string",
|
||||
"description": "Street address",
|
||||
},
|
||||
"apartmentNumber": {
|
||||
"type": "string",
|
||||
"description": "Apartment or unit number",
|
||||
},
|
||||
"zip": {"type": "string", "description": "Postal code"},
|
||||
"city": {"type": "string", "description": "City"},
|
||||
"country": {"type": "string", "description": "Country"},
|
||||
"email": {
|
||||
"type": "string",
|
||||
"description": "Email address (validated when provided)",
|
||||
},
|
||||
"phone": {
|
||||
"type": "string",
|
||||
"description": "Phone number in E.164 format (e.g., +14155551234)",
|
||||
},
|
||||
},
|
||||
"required": ["firstName"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="update_member",
|
||||
description="Update an existing member's information. All fields except id are optional.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {
|
||||
"type": "integer",
|
||||
"description": "The member's unique ID (required)",
|
||||
},
|
||||
"firstName": {
|
||||
"type": "string",
|
||||
"description": "Member's first name",
|
||||
},
|
||||
"lastName": {
|
||||
"type": "string",
|
||||
"description": "Member's last name",
|
||||
},
|
||||
"street": {
|
||||
"type": "string",
|
||||
"description": "Street address",
|
||||
},
|
||||
"apartmentNumber": {
|
||||
"type": "string",
|
||||
"description": "Apartment or unit number",
|
||||
},
|
||||
"zip": {"type": "string", "description": "Postal code"},
|
||||
"city": {"type": "string", "description": "City"},
|
||||
"country": {"type": "string", "description": "Country"},
|
||||
"email": {"type": "string", "description": "Email address"},
|
||||
"phone": {
|
||||
"type": "string",
|
||||
"description": "Phone number in E.164 format",
|
||||
},
|
||||
},
|
||||
"required": ["id"],
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
@self.server.call_tool()
|
||||
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
|
||||
"""Handle tool execution."""
|
||||
try:
|
||||
if name == "list_members":
|
||||
return await self._list_members()
|
||||
elif name == "get_member":
|
||||
return await self._get_member(arguments["id"])
|
||||
elif name == "create_member":
|
||||
return await self._create_member(arguments)
|
||||
elif name == "update_member":
|
||||
return await self._update_member(arguments)
|
||||
else:
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
except Exception as e:
|
||||
return [TextContent(type="text", text=f"Error: {str(e)}")]
|
||||
|
||||
async def _list_members(self) -> List[TextContent]:
|
||||
"""List all members."""
|
||||
query = """
|
||||
query {
|
||||
members {
|
||||
id
|
||||
firstName
|
||||
lastName
|
||||
street
|
||||
apartmentNumber
|
||||
zip
|
||||
city
|
||||
country
|
||||
email
|
||||
phone
|
||||
createdAt
|
||||
updatedAt
|
||||
}
|
||||
}
|
||||
"""
|
||||
data = await self.graphql.query(query)
|
||||
members = data.get("members", [])
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Found {len(members)} member(s):\n\n{self._format_members(members)}",
|
||||
)
|
||||
]
|
||||
|
||||
async def _get_member(self, member_id: int) -> List[TextContent]:
|
||||
"""Get a specific member by ID."""
|
||||
query = """
|
||||
query GetMember($id: Int!) {
|
||||
member(id: $id) {
|
||||
id
|
||||
firstName
|
||||
lastName
|
||||
street
|
||||
apartmentNumber
|
||||
zip
|
||||
city
|
||||
country
|
||||
email
|
||||
phone
|
||||
createdAt
|
||||
updatedAt
|
||||
}
|
||||
}
|
||||
"""
|
||||
data = await self.graphql.query(query, {"id": member_id})
|
||||
member = data.get("member")
|
||||
|
||||
if not member:
|
||||
return [
|
||||
TextContent(type="text", text=f"Member with ID {member_id} not found")
|
||||
]
|
||||
|
||||
return [
|
||||
TextContent(
|
||||
type="text", text=f"Member found:\n\n{self._format_member(member)}"
|
||||
)
|
||||
]
|
||||
|
||||
async def _create_member(self, input_data: Dict[str, Any]) -> List[TextContent]:
|
||||
"""Create a new member."""
|
||||
mutation = """
|
||||
mutation CreateMember($input: CreateMemberInput!) {
|
||||
createMember(input: $input) {
|
||||
id
|
||||
firstName
|
||||
lastName
|
||||
street
|
||||
apartmentNumber
|
||||
zip
|
||||
city
|
||||
country
|
||||
email
|
||||
phone
|
||||
createdAt
|
||||
updatedAt
|
||||
}
|
||||
}
|
||||
"""
|
||||
data = await self.graphql.query(mutation, {"input": input_data})
|
||||
member = data.get("createMember")
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Member created successfully:\n\n{self._format_member(member)}",
|
||||
)
|
||||
]
|
||||
|
||||
async def _update_member(self, input_data: Dict[str, Any]) -> List[TextContent]:
|
||||
"""Update an existing member."""
|
||||
mutation = """
|
||||
mutation UpdateMember($input: UpdateMemberInput!) {
|
||||
updateMember(input: $input) {
|
||||
id
|
||||
firstName
|
||||
lastName
|
||||
street
|
||||
apartmentNumber
|
||||
zip
|
||||
city
|
||||
country
|
||||
email
|
||||
phone
|
||||
createdAt
|
||||
updatedAt
|
||||
}
|
||||
}
|
||||
"""
|
||||
data = await self.graphql.query(mutation, {"input": input_data})
|
||||
member = data.get("updateMember")
|
||||
|
||||
if not member:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Failed to update member with ID {input_data.get('id')}",
|
||||
)
|
||||
]
|
||||
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Member updated successfully:\n\n{self._format_member(member)}",
|
||||
)
|
||||
]
|
||||
|
||||
def _format_member(self, member: Dict[str, Any]) -> str:
|
||||
"""Format a single member for display."""
|
||||
lines = [
|
||||
f"ID: {member.get('id')}",
|
||||
f"Name: {member.get('firstName', '')} {member.get('lastName', '')}".strip(),
|
||||
]
|
||||
|
||||
if member.get("street"):
|
||||
address_parts = [member.get("street")]
|
||||
if member.get("apartmentNumber"):
|
||||
address_parts.append(f"Apt {member.get('apartmentNumber')}")
|
||||
lines.append(f"Street: {' '.join(address_parts)}")
|
||||
|
||||
if member.get("zip") or member.get("city"):
|
||||
location = f"{member.get('zip', '')} {member.get('city', '')}".strip()
|
||||
lines.append(f"Location: {location}")
|
||||
|
||||
if member.get("country"):
|
||||
lines.append(f"Country: {member.get('country')}")
|
||||
|
||||
if member.get("email"):
|
||||
lines.append(f"Email: {member.get('email')}")
|
||||
|
||||
if member.get("phone"):
|
||||
lines.append(f"Phone: {member.get('phone')}")
|
||||
|
||||
if member.get("createdAt"):
|
||||
lines.append(f"Created: {member.get('createdAt')}")
|
||||
|
||||
if member.get("updatedAt"):
|
||||
lines.append(f"Updated: {member.get('updatedAt')}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def _format_members(self, members: List[Dict[str, Any]]) -> str:
|
||||
"""Format multiple members for display."""
|
||||
if not members:
|
||||
return "No members found."
|
||||
|
||||
return "\n\n---\n\n".join(self._format_member(m) for m in members)
|
||||
|
||||
async def run(self):
|
||||
"""Run the MCP server."""
|
||||
async with stdio_server() as (read_stream, write_stream):
|
||||
await self.server.run(
|
||||
read_stream,
|
||||
write_stream,
|
||||
self.server.create_initialization_options(),
|
||||
)
|
||||
|
||||
|
||||
async def main():
|
||||
"""Main entry point."""
|
||||
# Get API URL from environment or use default
|
||||
api_url = os.getenv("CLUBBER_API_URL", "http://127.0.0.1:8000/graphql")
|
||||
print("Starting Clubber MCP Server...", flush=True)
|
||||
print(f"GraphQL API URL: {api_url}", flush=True)
|
||||
|
||||
server = MemberManagementServer(api_url)
|
||||
try:
|
||||
await server.run()
|
||||
finally:
|
||||
await server.graphql.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user