nene2-python/src/example/mcp.py at main · hideyukiMORI/nene2-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
"""MCP server factory — registers Note, Tag, and Comment UseCases as MCP tools.

Run with:  uv run python -m example.mcp
Configure in Claude Desktop: see docs/howto/mcp-setup.md
"""

from dataclasses import asdict

from nene2.config import AppSettings
from nene2.mcp import LocalMcpServer

from .app import _build_repositories
from .comment.use_case import (
    CreateCommentInput,
    CreateCommentUseCase,
    DeleteCommentInput,
    DeleteCommentUseCase,
    GetCommentInput,
    GetCommentUseCase,
    ListCommentsInput,
    ListCommentsUseCase,
    UpdateCommentInput,
    UpdateCommentUseCase,
)
from .note.use_case import (
    CreateNoteInput,
    CreateNoteUseCase,
    DeleteNoteInput,
    DeleteNoteUseCase,
    GetNoteInput,
    GetNoteUseCase,
    ListNotesInput,
    ListNotesUseCase,
    UpdateNoteInput,
    UpdateNoteUseCase,
)
from .tag.use_case import (
    CreateTagInput,
    CreateTagUseCase,
    DeleteTagInput,
    DeleteTagUseCase,
    GetTagInput,
    GetTagUseCase,
    ListTagsInput,
    ListTagsUseCase,
    UpdateTagInput,
    UpdateTagUseCase,
)


def create_mcp_server(settings: AppSettings | None = None) -> LocalMcpServer:
    """Wire repositories and register all tools. Returns a runnable server."""
    cfg = settings or AppSettings()
    note_repo, tag_repo, comment_repo, _ = _build_repositories(cfg)

    note_list = ListNotesUseCase(note_repo)
    note_get = GetNoteUseCase(note_repo)
    note_create = CreateNoteUseCase(note_repo)
    note_update = UpdateNoteUseCase(note_repo)
    note_delete = DeleteNoteUseCase(note_repo)

    tag_list = ListTagsUseCase(tag_repo)
    tag_get = GetTagUseCase(tag_repo)
    tag_create = CreateTagUseCase(tag_repo)
    tag_update = UpdateTagUseCase(tag_repo)
    tag_delete = DeleteTagUseCase(tag_repo)

    comment_list = ListCommentsUseCase(comment_repo, note_repo)
    comment_get = GetCommentUseCase(comment_repo)
    comment_create = CreateCommentUseCase(comment_repo, note_repo)
    comment_update = UpdateCommentUseCase(comment_repo)
    comment_delete = DeleteCommentUseCase(comment_repo)

    server = LocalMcpServer(
        "nene2-example",
        instructions="Note, Tag, and Comment management API for the NENE2 example app.",
    )

    @server.tool("List notes with optional pagination.")
    def list_notes(limit: int = 20, offset: int = 0) -> list[dict]:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        result = note_list.execute(ListNotesInput(limit=limit, offset=offset))
        return [asdict(n) for n in result.items]

    @server.tool("Get a single note by ID.")
    def get_note(note_id: int) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(note_get.execute(GetNoteInput(note_id=note_id)))

    @server.tool("Create a new note.")
    def create_note(title: str, body: str) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(note_create.execute(CreateNoteInput(title=title, body=body)))

    @server.tool("Update an existing note.")
    def update_note(note_id: int, title: str, body: str) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(note_update.execute(UpdateNoteInput(note_id=note_id, title=title, body=body)))

    @server.tool("Delete a note by ID.")
    def delete_note(note_id: int) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        note_delete.execute(DeleteNoteInput(note_id=note_id))
        return {"deleted": True, "note_id": note_id}

    @server.tool("List tags with optional pagination.")
    def list_tags(limit: int = 20, offset: int = 0) -> list[dict]:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        result = tag_list.execute(ListTagsInput(limit=limit, offset=offset))
        return [asdict(t) for t in result.items]

    @server.tool("Get a single tag by ID.")
    def get_tag(tag_id: int) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(tag_get.execute(GetTagInput(tag_id=tag_id)))

    @server.tool("Create a new tag.")
    def create_tag(name: str) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(tag_create.execute(CreateTagInput(name=name)))

    @server.tool("Update an existing tag.")
    def update_tag(tag_id: int, name: str) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(tag_update.execute(UpdateTagInput(tag_id=tag_id, name=name)))

    @server.tool("Delete a tag by ID.")
    def delete_tag(tag_id: int) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        tag_delete.execute(DeleteTagInput(tag_id=tag_id))
        return {"deleted": True, "tag_id": tag_id}

    @server.tool("List comments for a note.")
    def list_comments(note_id: int, limit: int = 20, offset: int = 0) -> list[dict]:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        result = comment_list.execute(
            ListCommentsInput(note_id=note_id, limit=limit, offset=offset)
        )
        return [asdict(c) for c in result.items]

    @server.tool("Get a single comment by ID.")
    def get_comment(comment_id: int) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(comment_get.execute(GetCommentInput(comment_id=comment_id)))

    @server.tool("Create a new comment on a note.")
    def create_comment(note_id: int, body: str) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(comment_create.execute(CreateCommentInput(note_id=note_id, body=body)))

    @server.tool("Update an existing comment.")
    def update_comment(comment_id: int, body: str) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        return asdict(comment_update.execute(UpdateCommentInput(comment_id=comment_id, body=body)))

    @server.tool("Delete a comment by ID.")
    def delete_comment(comment_id: int) -> dict:  # type: ignore[type-arg]  # reason: mcp tool handler type stubs do not support generic dict
        comment_delete.execute(DeleteCommentInput(comment_id=comment_id))
        return {"deleted": True, "comment_id": comment_id}

    return server