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")