zzstoatzz.io
MCP server for tangled
1"""tangled MCP server - provides tools and resources for tangled git platform"""
2
3from typing import Annotated
4
5from fastmcp import FastMCP
6from pydantic import Field
7
8from tangled_mcp import _tangled
9from tangled_mcp.types import (
10 CreateIssueResult,
11 DeleteIssueResult,
12 ListBranchesResult,
13 ListIssuesResult,
14 ListPullsResult,
15 UpdateIssueResult,
16)
17
18tangled_mcp = FastMCP("tangled MCP server")
19
20
21# resources - read-only operations
22@tangled_mcp.resource("tangled://status")
23def tangled_status() -> dict[str, str | bool]:
24 """check the status of the tangled connection"""
25 client = _tangled._get_authenticated_client()
26
27 # verify can get tangled service token
28 try:
29 _tangled.get_service_token()
30 can_access_tangled = True
31 except Exception:
32 can_access_tangled = False
33
34 if not client.me:
35 raise RuntimeError("client not authenticated")
36
37 return {
38 "handle": client.me.handle,
39 "did": client.me.did,
40 "pds_authenticated": True,
41 "tangled_accessible": can_access_tangled,
42 }
43
44
45# tools - actions that query or modify state
46@tangled_mcp.tool
47def list_repo_branches(
48 repo: Annotated[
49 str,
50 Field(
51 description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')"
52 ),
53 ],
54 limit: Annotated[
55 int, Field(ge=1, le=100, description="maximum number of branches to return")
56 ] = 50,
57) -> ListBranchesResult:
58 """list branches for a repository
59
60 Args:
61 repo: repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')
62 limit: maximum number of branches to return (1-100)
63
64 Returns:
65 list of branches
66 """
67 # resolve owner/repo to (knot, did/repo)
68 knot, repo_id = _tangled.resolve_repo_identifier(repo)
69 response = _tangled.list_branches(knot, repo_id, limit, cursor=None)
70
71 return ListBranchesResult.from_api_response(response)
72
73
74@tangled_mcp.tool
75def create_repo_issue(
76 repo: Annotated[
77 str,
78 Field(
79 description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')"
80 ),
81 ],
82 title: Annotated[str, Field(description="issue title")],
83 body: Annotated[str | None, Field(description="issue body/description")] = None,
84 labels: Annotated[
85 list[str] | None,
86 Field(
87 description="optional list of label names (e.g., ['good-first-issue', 'bug']) "
88 "to apply to the issue"
89 ),
90 ] = None,
91) -> CreateIssueResult:
92 """create an issue on a repository
93
94 Args:
95 repo: repository identifier in 'owner/repo' format
96 title: issue title
97 body: optional issue body/description
98 labels: optional list of label names to apply
99
100 Returns:
101 CreateIssueResult with url (clickable link) and issue_id
102 """
103 # resolve owner/repo to (knot, did/repo)
104 knot, repo_id = _tangled.resolve_repo_identifier(repo)
105 # create_issue doesn't need knot (uses atproto putRecord, not XRPC)
106 response = _tangled.create_issue(repo_id, title, body, labels)
107
108 return CreateIssueResult(repo=repo, id=response["issueId"])
109
110
111@tangled_mcp.tool
112def update_repo_issue(
113 repo: Annotated[
114 str,
115 Field(
116 description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')"
117 ),
118 ],
119 issue_id: Annotated[int, Field(description="issue number (e.g., 1, 2, 3...)")],
120 title: Annotated[str | None, Field(description="new issue title")] = None,
121 body: Annotated[str | None, Field(description="new issue body/description")] = None,
122 labels: Annotated[
123 list[str] | None,
124 Field(
125 description="list of label names to SET (replaces all existing labels). "
126 "use empty list [] to remove all labels"
127 ),
128 ] = None,
129) -> UpdateIssueResult:
130 """update an existing issue on a repository
131
132 Args:
133 repo: repository identifier in 'owner/repo' format
134 issue_id: issue number to update
135 title: optional new title (if None, keeps existing)
136 body: optional new body (if None, keeps existing)
137 labels: optional list of label names to SET (replaces existing)
138
139 Returns:
140 UpdateIssueResult with url (clickable link) and issue_id
141 """
142 # resolve owner/repo to (knot, did/repo)
143 knot, repo_id = _tangled.resolve_repo_identifier(repo)
144 # update_issue doesn't need knot (uses atproto putRecord, not XRPC)
145 _tangled.update_issue(repo_id, issue_id, title, body, labels)
146
147 return UpdateIssueResult(repo=repo, id=issue_id)
148
149
150@tangled_mcp.tool
151def delete_repo_issue(
152 repo: Annotated[
153 str,
154 Field(
155 description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')"
156 ),
157 ],
158 issue_id: Annotated[
159 int, Field(description="issue number to delete (e.g., 1, 2, 3...)")
160 ],
161) -> DeleteIssueResult:
162 """delete an issue from a repository
163
164 Args:
165 repo: repository identifier in 'owner/repo' format
166 issue_id: issue number to delete
167
168 Returns:
169 DeleteIssueResult with issue_id of deleted issue
170 """
171 # resolve owner/repo to (knot, did/repo)
172 _, repo_id = _tangled.resolve_repo_identifier(repo)
173 # delete_issue doesn't need knot (uses atproto deleteRecord, not XRPC)
174 _tangled.delete_issue(repo_id, issue_id)
175
176 return DeleteIssueResult(id=issue_id)
177
178
179@tangled_mcp.tool
180def list_repo_issues(
181 repo: Annotated[
182 str,
183 Field(
184 description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')"
185 ),
186 ],
187 limit: Annotated[
188 int, Field(ge=1, le=100, description="maximum number of issues to return")
189 ] = 20,
190) -> ListIssuesResult:
191 """list issues for a repository
192
193 Args:
194 repo: repository identifier in 'owner/repo' format
195 limit: maximum number of issues to return (1-100)
196
197 Returns:
198 ListIssuesResult with list of issues
199 """
200 # resolve owner/repo to (knot, did/repo)
201 _, repo_id = _tangled.resolve_repo_identifier(repo)
202 # list_repo_issues doesn't need knot (queries atproto records, not XRPC)
203 response = _tangled.list_repo_issues(repo_id, limit, cursor=None)
204
205 return ListIssuesResult.from_api_response(response)
206
207
208@tangled_mcp.tool
209def list_repo_labels(
210 repo: Annotated[
211 str,
212 Field(
213 description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')"
214 ),
215 ],
216) -> list[str]:
217 """list available labels for a repository
218
219 Args:
220 repo: repository identifier in 'owner/repo' format
221
222 Returns:
223 list of available label names for the repository
224 """
225 # resolve owner/repo to (knot, did/repo)
226 _, repo_id = _tangled.resolve_repo_identifier(repo)
227 # list_repo_labels doesn't need knot (queries atproto records, not XRPC)
228 return _tangled.list_repo_labels(repo_id)
229
230
231@tangled_mcp.tool
232def list_repo_pulls(
233 repo: Annotated[
234 str,
235 Field(
236 description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')"
237 ),
238 ],
239 limit: Annotated[
240 int, Field(ge=1, le=100, description="maximum number of pulls to return")
241 ] = 20,
242) -> ListPullsResult:
243 """list pull requests created by the authenticated user for a repository
244
245 note: only returns PRs that the authenticated user created (tangled stores
246 PRs in the creator's repo, so we can only see our own PRs).
247
248 Args:
249 repo: repository identifier in 'owner/repo' format
250 limit: maximum number of pulls to return (1-100)
251
252 Returns:
253 ListPullsResult with list of pull requests
254 """
255 # resolve owner/repo to (knot, did/repo)
256 _, repo_id = _tangled.resolve_repo_identifier(repo)
257 # list_repo_pulls doesn't need knot (queries atproto records, not XRPC)
258 response = _tangled.list_repo_pulls(repo_id, limit)
259
260 return ListPullsResult.from_api_response(response["pulls"])