Tangled MCP Server

Instructions

Implementation Reference

• src/tangled_mcp/server.py:178-205 (handler)

  MCP tool handler: resolves repo identifier, calls core _tangled.list_repo_issues, returns ListIssuesResult instance.
  @tangled_mcp.tool def list_repo_issues( repo: Annotated[ str, Field( description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')" ), ], limit: Annotated[ int, Field(ge=1, le=100, description="maximum number of issues to return") ] = 20, ) -> ListIssuesResult: """list issues for a repository Args: repo: repository identifier in 'owner/repo' format limit: maximum number of issues to return (1-100) Returns: ListIssuesResult with list of issues """ # resolve owner/repo to (knot, did/repo) _, repo_id = _tangled.resolve_repo_identifier(repo) # list_repo_issues doesn't need knot (queries atproto records, not XRPC) response = _tangled.list_repo_issues(repo_id, limit, cursor=None) return ListIssuesResult.from_api_response(response)
• src/tangled_mcp/types/_issues.py:60-90 (schema)

  Pydantic model ListIssuesResult for output schema, with from_api_response classmethod to parse core API response.
  class ListIssuesResult(BaseModel): """result of listing issues""" issues: list[IssueInfo] @classmethod def from_api_response(cls, response: dict[str, Any]) -> "ListIssuesResult": """construct from raw API response Args: response: raw response from tangled API with structure: { "issues": [ { "uri": "at://...", "cid": "bafyrei...", "issueId": 1, "title": "...", "body": "...", "createdAt": "..." }, ... ] } Returns: ListIssuesResult with parsed issues """ issues = [IssueInfo(**issue_data) for issue_data in response.get("issues", [])] return cls(issues=issues)
• src/tangled_mcp/_tangled/_issues.py:328-434 (helper)

  Core helper function: queries atproto repo records for issues and label ops, filters by repo, assembles issue data with labels.
  def list_repo_issues( repo_id: str, limit: int = 50, cursor: str | None = None ) -> dict[str, Any]: """list issues for a repository Args: repo_id: repository identifier in "did/repo" format limit: maximum number of issues to return cursor: pagination cursor Returns: dict containing issues and optional cursor """ client = _get_authenticated_client() if not client.me: raise RuntimeError("client not authenticated") # parse repo_id to get owner_did and repo_name if "/" not in repo_id: raise ValueError(f"invalid repo_id format: {repo_id}") owner_did, repo_name = repo_id.split("/", 1) # get the repo AT-URI by querying the repo collection records = client.com.atproto.repo.list_records( models.ComAtprotoRepoListRecords.Params( repo=owner_did, collection="sh.tangled.repo", limit=100, ) ) repo_at_uri = None for record in records.records: if ( name := getattr(record.value, "name", None) ) is not None and name == repo_name: repo_at_uri = record.uri break if not repo_at_uri: raise ValueError(f"repo not found: {repo_id}") # list records from the issue collection response = client.com.atproto.repo.list_records( models.ComAtprotoRepoListRecords.Params( repo=client.me.did, collection="sh.tangled.repo.issue", limit=limit, cursor=cursor, ) ) # filter issues by repo issues = [] issue_uris = [] for record in response.records: if ( repo := getattr(record.value, "repo", None) ) is not None and repo == repo_at_uri: issue_uris.append(record.uri) issues.append( { "uri": record.uri, "cid": record.cid, "issueId": getattr(record.value, "issueId", 0), "title": getattr(record.value, "title", ""), "body": getattr(record.value, "body", None), "createdAt": getattr(record.value, "createdAt", ""), "labels": [], # will be populated below } ) # fetch label ops and correlate with issues if issue_uris: label_ops = client.com.atproto.repo.list_records( models.ComAtprotoRepoListRecords.Params( repo=client.me.did, collection="sh.tangled.label.op", limit=100, ) ) # build map of issue_uri -> current label URIs issue_labels_map: dict[str, set[str]] = {uri: set() for uri in issue_uris} for op_record in label_ops.records: if ( hasattr(op_record.value, "subject") and op_record.value.subject in issue_labels_map ): subject_uri = op_record.value.subject if hasattr(op_record.value, "add"): for operand in op_record.value.add: if hasattr(operand, "key"): issue_labels_map[subject_uri].add(operand.key) if hasattr(op_record.value, "delete"): for operand in op_record.value.delete: if hasattr(operand, "key"): issue_labels_map[subject_uri].discard(operand.key) # extract label names from URIs and add to issues for issue in issues: label_uris = issue_labels_map.get(issue["uri"], set()) issue["labels"] = [uri.split("/")[-1] for uri in label_uris] return {"issues": issues, "cursor": response.cursor}