MCP server for tangled
feat: add list_repo_pulls tool for reading pull requests
- add PullInfo, PullSource, PullTarget, ListPullsResult types
- implement list_repo_pulls in _tangled module
- note: only shows PRs created by authenticated user (atproto limitation)
- test: 7 tools now exposed
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+204
-1
+2
src/tangled_mcp/_tangled/__init__.py
+2
src/tangled_mcp/_tangled/__init__.py
···
13
13
list_repo_labels,
14
14
update_issue,
15
15
)
16
+
from tangled_mcp._tangled._pulls import list_repo_pulls
16
17
17
18
__all__ = [
18
19
"_get_authenticated_client",
···
23
24
"delete_issue",
24
25
"list_repo_issues",
25
26
"list_repo_labels",
27
+
"list_repo_pulls",
26
28
"resolve_repo_identifier",
27
29
]
+92
src/tangled_mcp/_tangled/_pulls.py
+92
src/tangled_mcp/_tangled/_pulls.py
···
1
+
"""pull request operations for tangled"""
2
+
3
+
from typing import Any
4
+
5
+
from atproto import models
6
+
7
+
from tangled_mcp._tangled._client import _get_authenticated_client
8
+
9
+
10
+
def list_repo_pulls(repo_id: str, limit: int = 50) -> dict[str, Any]:
11
+
"""list pull requests created by the authenticated user for a repository
12
+
13
+
note: this only returns PRs that the authenticated user created.
14
+
tangled stores PRs in the creator's repo, so we can only see our own PRs.
15
+
16
+
Args:
17
+
repo_id: repository identifier in "did/repo" format
18
+
limit: maximum number of pulls to return
19
+
20
+
Returns:
21
+
dict containing pulls list
22
+
"""
23
+
client = _get_authenticated_client()
24
+
25
+
if not client.me:
26
+
raise RuntimeError("client not authenticated")
27
+
28
+
# parse repo_id to get owner_did and repo_name
29
+
if "/" not in repo_id:
30
+
raise ValueError(f"invalid repo_id format: {repo_id}")
31
+
32
+
owner_did, repo_name = repo_id.split("/", 1)
33
+
34
+
# get the repo AT-URI by querying the repo collection
35
+
records = client.com.atproto.repo.list_records(
36
+
models.ComAtprotoRepoListRecords.Params(
37
+
repo=owner_did,
38
+
collection="sh.tangled.repo",
39
+
limit=100,
40
+
)
41
+
)
42
+
43
+
repo_at_uri = None
44
+
for record in records.records:
45
+
if (name := getattr(record.value, "name", None)) is not None and name == repo_name:
46
+
repo_at_uri = record.uri
47
+
break
48
+
49
+
if not repo_at_uri:
50
+
raise ValueError(f"repo not found: {repo_id}")
51
+
52
+
# list pull records from the authenticated user's collection
53
+
response = client.com.atproto.repo.list_records(
54
+
models.ComAtprotoRepoListRecords.Params(
55
+
repo=client.me.did,
56
+
collection="sh.tangled.repo.pull",
57
+
limit=limit,
58
+
)
59
+
)
60
+
61
+
# filter pulls by target repo and convert to dict format
62
+
pulls = []
63
+
for record in response.records:
64
+
value = record.value
65
+
target = getattr(value, "target", None)
66
+
if not target:
67
+
continue
68
+
69
+
target_repo = getattr(target, "repo", None)
70
+
if target_repo != repo_at_uri:
71
+
continue
72
+
73
+
source = getattr(value, "source", {})
74
+
pulls.append(
75
+
{
76
+
"uri": record.uri,
77
+
"cid": record.cid,
78
+
"title": getattr(value, "title", ""),
79
+
"source": {
80
+
"sha": getattr(source, "sha", ""),
81
+
"branch": getattr(source, "branch", ""),
82
+
"repo": getattr(source, "repo", None),
83
+
},
84
+
"target": {
85
+
"repo": target_repo,
86
+
"branch": getattr(target, "branch", ""),
87
+
},
88
+
"createdAt": getattr(value, "createdAt", ""),
89
+
}
90
+
)
91
+
92
+
return {"pulls": pulls}
+33
src/tangled_mcp/server.py
+33
src/tangled_mcp/server.py
···
11
11
DeleteIssueResult,
12
12
ListBranchesResult,
13
13
ListIssuesResult,
14
+
ListPullsResult,
14
15
UpdateIssueResult,
15
16
)
16
17
···
225
226
_, repo_id = _tangled.resolve_repo_identifier(repo)
226
227
# list_repo_labels doesn't need knot (queries atproto records, not XRPC)
227
228
return _tangled.list_repo_labels(repo_id)
229
+
230
+
231
+
@tangled_mcp.tool
232
+
def 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"])
+5
src/tangled_mcp/types/__init__.py
+5
src/tangled_mcp/types/__init__.py
···
9
9
ListIssuesResult,
10
10
UpdateIssueResult,
11
11
)
12
+
from tangled_mcp.types._pulls import ListPullsResult, PullInfo, PullSource, PullTarget
12
13
13
14
__all__ = [
14
15
"BranchInfo",
···
17
18
"IssueInfo",
18
19
"ListBranchesResult",
19
20
"ListIssuesResult",
21
+
"ListPullsResult",
22
+
"PullInfo",
23
+
"PullSource",
24
+
"PullTarget",
20
25
"RepoIdentifier",
21
26
"UpdateIssueResult",
22
27
]
+70
src/tangled_mcp/types/_pulls.py
+70
src/tangled_mcp/types/_pulls.py
···
1
+
"""pull request types"""
2
+
3
+
from typing import Any
4
+
5
+
from pydantic import BaseModel, Field
6
+
7
+
8
+
class PullSource(BaseModel):
9
+
"""source branch info for a pull request"""
10
+
11
+
sha: str
12
+
branch: str
13
+
repo: str | None = None # AT-URI of source repo (for cross-repo PRs)
14
+
15
+
16
+
class PullTarget(BaseModel):
17
+
"""target branch info for a pull request"""
18
+
19
+
repo: str # AT-URI of target repo
20
+
branch: str
21
+
22
+
23
+
class PullInfo(BaseModel):
24
+
"""pull request information"""
25
+
26
+
uri: str
27
+
cid: str
28
+
title: str
29
+
source: PullSource
30
+
target: PullTarget
31
+
created_at: str = Field(alias="createdAt")
32
+
33
+
34
+
class ListPullsResult(BaseModel):
35
+
"""result of listing pull requests"""
36
+
37
+
pulls: list[PullInfo]
38
+
39
+
@classmethod
40
+
def from_api_response(cls, pulls_data: list[dict[str, Any]]) -> "ListPullsResult":
41
+
"""construct from pre-filtered pull data
42
+
43
+
Args:
44
+
pulls_data: list of pull dicts already filtered by target repo
45
+
46
+
Returns:
47
+
ListPullsResult with parsed pulls
48
+
"""
49
+
pulls = []
50
+
for pull in pulls_data:
51
+
source = pull.get("source", {})
52
+
target = pull.get("target", {})
53
+
pulls.append(
54
+
PullInfo(
55
+
uri=pull["uri"],
56
+
cid=pull["cid"],
57
+
title=pull.get("title", ""),
58
+
source=PullSource(
59
+
sha=source.get("sha", ""),
60
+
branch=source.get("branch", ""),
61
+
repo=source.get("repo"),
62
+
),
63
+
target=PullTarget(
64
+
repo=target.get("repo", ""),
65
+
branch=target.get("branch", ""),
66
+
),
67
+
createdAt=pull.get("createdAt", ""),
68
+
)
69
+
)
70
+
return cls(pulls=pulls)
+2
-1
tests/test_server.py
+2
-1
tests/test_server.py
···
22
22
async with Client(tangled_mcp) as client:
23
23
tools = await client.list_tools()
24
24
25
-
assert len(tools) == 6
25
+
assert len(tools) == 7
26
26
27
27
tool_names = {tool.name for tool in tools}
28
28
assert "list_repo_branches" in tool_names
···
31
31
assert "delete_repo_issue" in tool_names
32
32
assert "list_repo_issues" in tool_names
33
33
assert "list_repo_labels" in tool_names
34
+
assert "list_repo_pulls" in tool_names
34
35
35
36
async def test_list_repo_branches_tool_schema(self):
36
37
"""test list_repo_branches tool has correct schema"""