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}