Tangled MCP Server

Instructions

Implementation Reference

• src/tangled_mcp/server.py:110-146 (handler)

  Core handler function for the 'update_repo_issue' tool, decorated with @tangled_mcp.tool for registration. Includes input schema via Annotated Fields and Pydantic return type. Executes repo resolution and delegates to _tangled.update_issue helper.

  @tangled_mcp.tool
  def update_repo_issue(
      repo: Annotated[
          str,
          Field(
              description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')"
          ),
      ],
      issue_id: Annotated[int, Field(description="issue number (e.g., 1, 2, 3...)")],
      title: Annotated[str | None, Field(description="new issue title")] = None,
      body: Annotated[str | None, Field(description="new issue body/description")] = None,
      labels: Annotated[
          list[str] | None,
          Field(
              description="list of label names to SET (replaces all existing labels). "
              "use empty list [] to remove all labels"
          ),
      ] = None,
  ) -> UpdateIssueResult:
      """update an existing issue on a repository
  
      Args:
          repo: repository identifier in 'owner/repo' format
          issue_id: issue number to update
          title: optional new title (if None, keeps existing)
          body: optional new body (if None, keeps existing)
          labels: optional list of label names to SET (replaces existing)
  
      Returns:
          UpdateIssueResult with url (clickable link) and issue_id
      """
      # resolve owner/repo to (knot, did/repo)
      knot, repo_id = _tangled.resolve_repo_identifier(repo)
      # update_issue doesn't need knot (uses atproto putRecord, not XRPC)
      _tangled.update_issue(repo_id, issue_id, title, body, labels)
  
      return UpdateIssueResult(repo=repo, issue_id=issue_id)

• src/tangled_mcp/types/_issues.py:41-51 (schema)

  Pydantic model defining the output schema for the update_repo_issue tool, including computed URL field.

  class UpdateIssueResult(BaseModel):
      """result of updating an issue"""
  
      repo: RepoIdentifier
      issue_id: int
  
      @computed_field
      @property
      def url(self) -> str:
          """construct clickable tangled.org URL"""
          return _tangled_issue_url(self.repo, self.issue_id)

• src/tangled_mcp/_tangled/_issues.py:121-247 (helper)

  Helper function implementing the core update logic: locates repo and issue records via atproto list_records, constructs updated record preserving unchanged fields, performs put_record with swap for atomic update, and handles label changes via label op records.

  def update_issue(
      repo_id: str,
      issue_id: int,
      title: str | None = None,
      body: str | None = None,
      labels: list[str] | None = None,
  ) -> dict[str, Any]:
      """update an existing issue on a repository
  
      Args:
          repo_id: repository identifier in "did/repo" format (e.g., 'did:plc:.../tangled-mcp')
          issue_id: the sequential issue number (e.g., 1, 2, 3...)
          title: optional new issue title (if None, keeps existing)
          body: optional new issue body (if None, keeps existing)
          labels: optional list of label names to SET (replaces all existing labels)
                  use empty list [] to remove all labels
  
      Returns:
          dict with uri and cid of updated issue record
      """
      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 and label definitions
      records = client.com.atproto.repo.list_records(
          models.ComAtprotoRepoListRecords.Params(
              repo=owner_did,
              collection="sh.tangled.repo",
              limit=100,
          )
      )
  
      repo_at_uri = None
      repo_labels: list[str] = []
      for record in records.records:
          if (name := getattr(record.value, "name", None)) and name == repo_name:
              repo_at_uri = record.uri
              if (subscribed_labels := getattr(record.value, "labels", None)) is not None:
                  repo_labels = subscribed_labels
              break
  
      if not repo_at_uri:
          raise ValueError(f"repo not found: {repo_id}")
  
      # find the issue record with matching issueId
      existing_issues = client.com.atproto.repo.list_records(
          models.ComAtprotoRepoListRecords.Params(
              repo=client.me.did,
              collection="sh.tangled.repo.issue",
              limit=100,
          )
      )
  
      issue_record = None
      issue_rkey = None
      for record in existing_issues.records:
          if (
              (repo := getattr(record.value, "repo", None)) is not None
              and repo == repo_at_uri
              and (_issue_id := getattr(record.value, "issueId", None)) is not None
              and _issue_id == issue_id
          ):
              issue_record = record
              issue_rkey = record.uri.split("/")[-1]  # extract rkey from AT-URI
              break
  
      if not issue_record:
          raise ValueError(f"issue #{issue_id} not found in repo {repo_id}")
  
      # update the issue fields (keep existing if not specified)
      updated_record = {
          "$type": "sh.tangled.repo.issue",
          "repo": repo_at_uri,
          "issueId": issue_id,
          "owner": (
              (owner := getattr(issue_record.value, "owner", None)) is not None and owner
              if hasattr(issue_record.value, "owner")
              else client.me.did
          ),
          "title": title
          if title is not None
          else getattr(issue_record.value, "title", None),
          "body": body if body is not None else getattr(issue_record.value, "body", None),
          "createdAt": getattr(issue_record.value, "createdAt", None),
      }
  
      # get current CID for swap
      current_cid = issue_record.cid
  
      # update the issue record
  
      if issue_rkey is None:
          raise ValueError(
              f"issue rkey not found for issue #{issue_id} in repo {repo_id}"
          )
  
      response = client.com.atproto.repo.put_record(
          models.ComAtprotoRepoPutRecord.Data(
              repo=client.me.did,
              collection="sh.tangled.repo.issue",
              rkey=issue_rkey,
              record=updated_record,
              swap_record=current_cid,  # ensure we're updating the right version
          )
      )
  
      result = {"uri": response.uri, "cid": response.cid}
  
      # if labels were specified, create a label op to set them
      if labels is not None:
          issue_uri = response.uri
  
          # get current label state for this issue
          current_labels = _get_current_labels(client, issue_uri)
  
          # apply the new label state
          _apply_labels(client, issue_uri, labels, repo_labels, current_labels)
  
      return result

• src/tangled_mcp/server.py:17-17 (registration)

  FastMCP server instance creation where all tools are registered via decorators. The update_repo_issue tool is registered via @tangled_mcp.tool decorator on its handler.

  tangled_mcp = FastMCP("tangled MCP server")