GraphQL MCP Examples

"""Hello World - minimal GraphQL MCP server example."""

from graphql_api import GraphQLAPI, field
from graphql_mcp.server import GraphQLMCP


class HelloWorldAPI:

    @field
    def hello(self, name: str = "World") -> str:
        """Say hello to someone."""
        return f"Hello, {name}!"


api = GraphQLAPI(root_type=HelloWorldAPI)
server = GraphQLMCP.from_api(api, graphql_http_kwargs={
    "graphiql_example_query": """\
{
  hello

  custom: hello(name: "GraphQL MCP")
}""",
})
app = server.http_app(transport="streamable-http", stateless_http=True)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8002)

"""Task Manager - CRUD example with enums, mutations, and in-memory state.

Demonstrates:
- Dataclass types as GraphQL object types
- Enums (Priority, Status)
- Queries with optional filters
- Mutations via @field(mutable=True)
- UUID and datetime scalars
- Optional fields and list types
"""

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Optional
from uuid import UUID, uuid4

from graphql_api import GraphQLAPI, field
from graphql_mcp.server import GraphQLMCP


class Priority(str, Enum):
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"


class Status(str, Enum):
    TODO = "TODO"
    IN_PROGRESS = "IN_PROGRESS"
    DONE = "DONE"


@dataclass
class Task:
    id: UUID
    title: str
    description: str
    status: Status
    priority: Priority
    tags: list[str]
    created_at: datetime
    completed_at: Optional[datetime] = None


# In-memory store, seeded with sample data
_tasks: dict[UUID, Task] = {}


def _seed():
    for title, desc, priority, status, tags in [
        ("Set up CI/CD", "Configure GitHub Actions pipeline",
         Priority.HIGH, Status.DONE, ["devops", "infrastructure"]),
        ("Write API docs", "Document all GraphQL endpoints",
         Priority.MEDIUM, Status.IN_PROGRESS, ["docs"]),
        ("Fix login bug", "Users getting 401 on valid credentials",
         Priority.CRITICAL, Status.TODO, ["bug", "auth"]),
        ("Add dark mode", "Support dark theme in web app",
         Priority.LOW, Status.TODO, ["frontend", "ui"]),
    ]:
        task = Task(
            id=uuid4(), title=title, description=desc,
            status=status, priority=priority, tags=tags,
            created_at=datetime(2026, 1, 15, 9, 0, 0),
            completed_at=datetime(2026, 2, 1, 14, 30, 0) if status == Status.DONE else None,
        )
        _tasks[task.id] = task


_seed()


class TaskManagerAPI:

    @field
    def tasks(
        self,
        status: Optional[Status] = None,
        priority: Optional[Priority] = None,
    ) -> list[Task]:
        """List all tasks, optionally filtered by status or priority."""
        result = list(_tasks.values())
        if status is not None:
            result = [t for t in result if t.status == status]
        if priority is not None:
            result = [t for t in result if t.priority == priority]
        return result

    @field
    def task(self, id: UUID) -> Optional[Task]:
        """Get a single task by ID."""
        return _tasks.get(id)

    @field(mutable=True)
    def create_task(
        self,
        title: str,
        description: str = "",
        priority: Priority = Priority.MEDIUM,
        tags: Optional[list[str]] = None,
    ) -> Task:
        """Create a new task."""
        task = Task(
            id=uuid4(),
            title=title,
            description=description,
            status=Status.TODO,
            priority=priority,
            tags=tags or [],
            created_at=datetime.now(),
        )
        _tasks[task.id] = task
        return task

    @field(mutable=True)
    def update_status(self, id: UUID, status: Status) -> Task:
        """Update a task's status. Automatically sets completed_at when done."""
        task = _tasks[id]
        task.status = status
        if status == Status.DONE:
            task.completed_at = datetime.now()
        elif task.completed_at is not None:
            task.completed_at = None
        return task

    @field(mutable=True)
    def delete_task(self, id: UUID) -> bool:
        """Delete a task by ID. Returns true if the task existed."""
        if id in _tasks:
            del _tasks[id]
            return True
        return False


api = GraphQLAPI(root_type=TaskManagerAPI)
server = GraphQLMCP.from_api(api, allow_mutations=True, graphql_http_kwargs={
    "graphiql_example_query": """\
# List all tasks
{
  tasks {
    id
    title
    status
    priority
    tags
    createdAt
  }
}

# Filter by status
# {
#   tasks(status: TODO) {
#     title
#     priority
#   }
# }

# Create a task
# mutation {
#   createTask(
#     title: "My new task"
#     description: "Something important"
#     priority: HIGH
#     tags: ["example"]
#   ) {
#     id
#     title
#     createdAt
#   }
# }""",
})
app = server.http_app(transport="streamable-http", stateless_http=True)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8003)

"""Nested API - demonstrates nested tools, @mcpHidden, Pydantic models, and async resolvers.

Demonstrates:
- Nested query paths that auto-generate MCP tools (category → articles)
- @mcpHidden directive to hide arguments from MCP tools
- Pydantic BaseModel types as GraphQL object types
- Async resolvers
- Separate query_type / mutation_type pattern
"""

from typing import Annotated, Optional
from uuid import uuid4

from pydantic import BaseModel

from graphql_api import GraphQLAPI, field
from graphql_mcp import GraphQLMCP, mcp_hidden


class Comment(BaseModel):
    id: str
    author: str
    text: str


class Article(BaseModel):
    id: str
    title: str
    body: str
    tags: list[str]
    comments: list[Comment] = []


# In-memory store keyed by category name
_categories: dict[str, list[Article]] = {}


def _seed():
    _categories["python"] = [
        Article(
            id=str(uuid4()), title="Getting Started with FastAPI",
            body="FastAPI is a modern web framework for building APIs with Python.",
            tags=["web", "fastapi"],
            comments=[
                Comment(id=str(uuid4()), author="alice", text="Great intro!"),
                Comment(id=str(uuid4()), author="bob", text="Very helpful."),
            ],
        ),
        Article(
            id=str(uuid4()), title="Async Python Patterns",
            body="Learn how to use asyncio effectively in your projects.",
            tags=["async", "patterns"],
            comments=[
                Comment(id=str(uuid4()), author="charlie", text="Exactly what I needed."),
            ],
        ),
    ]
    _categories["graphql"] = [
        Article(
            id=str(uuid4()), title="Schema-First vs Code-First GraphQL",
            body="Comparing the two main approaches to building GraphQL APIs.",
            tags=["architecture", "patterns"],
            comments=[],
        ),
        Article(
            id=str(uuid4()), title="GraphQL Subscriptions Deep Dive",
            body="Understanding real-time data with GraphQL subscriptions.",
            tags=["real-time", "subscriptions"],
            comments=[
                Comment(id=str(uuid4()), author="diana", text="When would you use SSE instead?"),
            ],
        ),
    ]
    _categories["mcp"] = [
        Article(
            id=str(uuid4()), title="Building MCP Servers",
            body="How to expose your API as MCP tools for AI agents.",
            tags=["ai", "mcp"],
            comments=[
                Comment(id=str(uuid4()), author="eve", text="This is the future."),
            ],
        ),
    ]


_seed()


class Category:
    """A category containing articles. Fields with arguments generate nested MCP tools."""

    def __init__(self, name: str, articles: list[Article]):
        self._name = name
        self._articles = articles

    @field
    async def articles(
        self,
        tag: Optional[str] = None,
        internal_score: Annotated[Optional[int], mcp_hidden] = None,
    ) -> list[Article]:
        """List articles in this category, optionally filtered by tag.

        The internal_score argument is hidden from MCP tools via @mcpHidden
        but remains accessible through the GraphQL API directly.
        """
        result = self._articles
        if tag is not None:
            result = [a for a in result if tag in a.tags]
        return result

    @field
    def name(self) -> str:
        """The category name."""
        return self._name

    @field
    def article_count(self) -> int:
        """Number of articles in this category."""
        return len(self._articles)


class Query:

    @field
    def categories(self) -> list[str]:
        """List all category names."""
        return list(_categories.keys())

    @field
    def category(self, name: str) -> Optional[Category]:
        """Get a category by name."""
        articles = _categories.get(name)
        if articles is None:
            return None
        return Category(name, articles)


class Mutation:

    @field(mutable=True)
    def add_article(
        self,
        category: str,
        title: str,
        body: str,
        tags: Optional[list[str]] = None,
    ) -> Article:
        """Add an article to a category. Creates the category if it doesn't exist."""
        article = Article(
            id=str(uuid4()),
            title=title,
            body=body,
            tags=tags or [],
        )
        if category not in _categories:
            _categories[category] = []
        _categories[category].append(article)
        return article


api = GraphQLAPI(
    query_type=Query,
    mutation_type=Mutation,
    directives=[mcp_hidden],
)
server = GraphQLMCP.from_api(api, allow_mutations=True, graphql_http_kwargs={
    "graphiql_example_query": """\
# Browse the knowledge base
{
  categories

  category(name: "python") {
    name
    articleCount
    articles {
      title
      tags
      comments {
        author
        text
      }
    }
  }
}

# Filter articles by tag
# {
#   category(name: "python") {
#     articles(tag: "async") {
#       title
#       body
#     }
#   }
# }

# Add an article
# mutation {
#   addArticle(
#     category: "python"
#     title: "My New Article"
#     body: "Article content here"
#     tags: ["tutorial"]
#   ) {
#     id
#     title
#   }
# }""",
})
app = server.http_app(transport="streamable-http", stateless_http=True)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8005)

"""Remote API - wraps a public GraphQL API as MCP tools.

Demonstrates:
- GraphQLMCP.from_remote_url() to introspect and wrap any GraphQL endpoint
- Auto-generated MCP tools from a remote schema
- Read-only access (allow_mutations=False)

Uses the public Countries GraphQL API (https://countries.trevorblades.com).
"""

from graphql_mcp.server import GraphQLMCP

server = GraphQLMCP.from_remote_url(
    "https://countries.trevorblades.com/graphql",
    allow_mutations=False,
    graphql_http_kwargs={
        "graphiql_example_query": """\
{
  countries(filter: { continent: { eq: "EU" } }) {
    name
    capital
    emoji
    languages {
      name
    }
  }
}""",
    },
)
app = server.http_app(transport="streamable-http", stateless_http=True)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8004)