list_repo_pulls

Instructions

Implementation Reference

• src/tangled_mcp/_tangled/_pulls.py:10-92 (handler)

  Core implementation of list_repo_pulls — queries atproto records (sh.tangled.repo.pull) filtered by target repo. Returns dict with pulls list.

  def list_repo_pulls(repo_id: str, limit: int = 50) -> dict[str, Any]:
      """list pull requests created by the authenticated user for a repository
  
      note: this only returns PRs that the authenticated user created.
      tangled stores PRs in the creator's repo, so we can only see our own PRs.
  
      Args:
          repo_id: repository identifier in "did/repo" format
          limit: maximum number of pulls to return
  
      Returns:
          dict containing pulls list
      """
      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 pull records from the authenticated user's collection
      response = client.com.atproto.repo.list_records(
          models.ComAtprotoRepoListRecords.Params(
              repo=client.me.did,
              collection="sh.tangled.repo.pull",
              limit=limit,
          )
      )
  
      # filter pulls by target repo and convert to dict format
      pulls = []
      for record in response.records:
          value = record.value
          target = getattr(value, "target", None)
          if not target:
              continue
  
          target_repo = getattr(target, "repo", None)
          if target_repo != repo_at_uri:
              continue
  
          source = getattr(value, "source", {})
          pulls.append(
              {
                  "uri": record.uri,
                  "cid": record.cid,
                  "title": getattr(value, "title", ""),
                  "source": {
                      "sha": getattr(source, "sha", ""),
                      "branch": getattr(source, "branch", ""),
                      "repo": getattr(source, "repo", None),
                  },
                  "target": {
                      "repo": target_repo,
                      "branch": getattr(target, "branch", ""),
                  },
                  "createdAt": getattr(value, "createdAt", ""),
              }
          )
  
      return {"pulls": pulls}

• src/tangled_mcp/server.py:231-261 (handler)

  MCP tool handler decorated with @tangled_mcp.tool. Resolves repo identifier, delegates to _tangled.list_repo_pulls, returns ListPullsResult.

  @tangled_mcp.tool
  def list_repo_pulls(
      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 pulls to return")
      ] = 20,
  ) -> ListPullsResult:
      """list pull requests created by the authenticated user for a repository
  
      note: only returns PRs that the authenticated user created (tangled stores
      PRs in the creator's repo, so we can only see our own PRs).
  
      Args:
          repo: repository identifier in 'owner/repo' format
          limit: maximum number of pulls to return (1-100)
  
      Returns:
          ListPullsResult with list of pull requests
      """
      # resolve owner/repo to (knot, did/repo)
      _, repo_id = _tangled.resolve_repo_identifier(repo)
      # list_repo_pulls doesn't need knot (queries atproto records, not XRPC)
      response = _tangled.list_repo_pulls(repo_id, limit)
  
      return ListPullsResult.from_api_response(response["pulls"])

• src/tangled_mcp/types/_pulls.py:1-70 (schema)

  Schema types: PullSource, PullTarget, PullInfo, ListPullsResult with from_api_response classmethod.

  """pull request types"""
  
  from typing import Any
  
  from pydantic import BaseModel, Field
  
  
  class PullSource(BaseModel):
      """source branch info for a pull request"""
  
      sha: str
      branch: str
      repo: str | None = None  # AT-URI of source repo (for cross-repo PRs)
  
  
  class PullTarget(BaseModel):
      """target branch info for a pull request"""
  
      repo: str  # AT-URI of target repo
      branch: str
  
  
  class PullInfo(BaseModel):
      """pull request information"""
  
      uri: str
      cid: str
      title: str
      source: PullSource
      target: PullTarget
      created_at: str = Field(alias="createdAt")
  
  
  class ListPullsResult(BaseModel):
      """result of listing pull requests"""
  
      pulls: list[PullInfo]
  
      @classmethod
      def from_api_response(cls, pulls_data: list[dict[str, Any]]) -> "ListPullsResult":
          """construct from pre-filtered pull data
  
          Args:
              pulls_data: list of pull dicts already filtered by target repo
  
          Returns:
              ListPullsResult with parsed pulls
          """
          pulls = []
          for pull in pulls_data:
              source = pull.get("source", {})
              target = pull.get("target", {})
              pulls.append(
                  PullInfo(
                      uri=pull["uri"],
                      cid=pull["cid"],
                      title=pull.get("title", ""),
                      source=PullSource(
                          sha=source.get("sha", ""),
                          branch=source.get("branch", ""),
                          repo=source.get("repo"),
                      ),
                      target=PullTarget(
                          repo=target.get("repo", ""),
                          branch=target.get("branch", ""),
                      ),
                      createdAt=pull.get("createdAt", ""),
                  )
              )
          return cls(pulls=pulls)

• src/tangled_mcp/_tangled/__init__.py:16-27 (registration)

  Registration: re-exports list_repo_pulls from _pulls module in the _tangled package's public API.

  from tangled_mcp._tangled._pulls import list_repo_pulls
  
  __all__ = [
      "_get_authenticated_client",
      "get_service_token",
      "list_branches",
      "create_issue",
      "update_issue",
      "delete_issue",
      "list_repo_issues",
      "list_repo_labels",
      "list_repo_pulls",

• src/tangled_mcp/types/__init__.py:12-12 (registration)

  Registration: re-exports ListPullsResult (and PullInfo, PullSource, PullTarget) from types package.

  from tangled_mcp.types._pulls import ListPullsResult, PullInfo, PullSource, PullTarget