Tangled MCP Server

Instructions

Implementation Reference

• src/tangled_mcp/server.py:73-108 (handler)

  The primary MCP tool handler for 'create_repo_issue'. This function defines the tool schema via Annotated parameters and Pydantic Fields, registers it via @tangled_mcp.tool decorator, resolves the repo identifier, delegates to the core _tangled.create_issue helper, and returns a typed CreateIssueResult.
  @tangled_mcp.tool def create_repo_issue( repo: Annotated[ str, Field( description="repository identifier in 'owner/repo' format (e.g., 'zzstoatzz/tangled-mcp')" ), ], title: Annotated[str, Field(description="issue title")], body: Annotated[str | None, Field(description="issue body/description")] = None, labels: Annotated[ list[str] | None, Field( description="optional list of label names (e.g., ['good-first-issue', 'bug']) " "to apply to the issue" ), ] = None, ) -> CreateIssueResult: """create an issue on a repository Args: repo: repository identifier in 'owner/repo' format title: issue title body: optional issue body/description labels: optional list of label names to apply Returns: CreateIssueResult with url (clickable link) and issue_id """ # resolve owner/repo to (knot, did/repo) knot, repo_id = _tangled.resolve_repo_identifier(repo) # create_issue doesn't need knot (uses atproto putRecord, not XRPC) response = _tangled.create_issue(repo_id, title, body, labels) return CreateIssueResult(repo=repo, issue_id=response["issueId"])
• src/tangled_mcp/types/_issues.py:28-39 (schema)

  Pydantic model defining the output schema for the create_repo_issue tool, including a computed 'url' property based on repo and issue_id.
  class CreateIssueResult(BaseModel): """result of creating 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:11-118 (helper)

  Core helper function implementing the issue creation logic using the atproto client. Handles repo resolution from did/repo, next issue ID calculation, record creation via put_record, label validation and application.
  def create_issue( repo_id: str, title: str, body: str | None = None, labels: list[str] | None = None ) -> dict[str, Any]: """create an issue on a repository Args: repo_id: repository identifier in "did/repo" format (e.g., 'did:plc:.../tangled-mcp') title: issue title body: optional issue body/description labels: optional list of label names (e.g., ["good-first-issue", "bug"]) or full label definition URIs (e.g., ["at://did:.../sh.tangled.label.definition/bug"]) Returns: dict with uri, cid, and issueId of created 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 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 repo_labels: list[str] = [] for record in records.records: if ( name := getattr(record.value, "name", None) ) is not None and name == repo_name: repo_at_uri = record.uri # get repo's subscribed labels 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}") # query existing issues to determine next issueId existing_issues = client.com.atproto.repo.list_records( models.ComAtprotoRepoListRecords.Params( repo=client.me.did, collection="sh.tangled.repo.issue", limit=100, ) ) # find max issueId for this repo max_issue_id = 0 for issue_record in existing_issues.records: if ( repo := getattr(issue_record.value, "repo", None) ) is not None and repo == repo_at_uri: issue_id = getattr(issue_record.value, "issueId", None) if issue_id is not None: max_issue_id = max(max_issue_id, issue_id) next_issue_id = max_issue_id + 1 # validate labels BEFORE creating the issue to prevent orphaned issues if labels: _validate_labels(labels, repo_labels) # generate timestamp ID for rkey tid = int(datetime.now(timezone.utc).timestamp() * 1000000) rkey = str(tid) # create issue record with proper schema record = { "$type": "sh.tangled.repo.issue", "repo": repo_at_uri, # full AT-URI of repo record "issueId": next_issue_id, # sequential issue ID "owner": client.me.did, # issue creator's DID "title": title, "body": body, "createdAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), } # use putRecord to create the issue response = client.com.atproto.repo.put_record( models.ComAtprotoRepoPutRecord.Data( repo=client.me.did, collection="sh.tangled.repo.issue", rkey=rkey, record=record, ) ) issue_uri = response.uri result = {"uri": issue_uri, "cid": response.cid, "issueId": next_issue_id} # if labels were specified, create a label op to add them if labels: _apply_labels(client, issue_uri, labels, repo_labels, current_labels=set()) return result
• src/tangled_mcp/_tangled/_client.py:11-69 (helper)

  Helper function called by the handler to resolve 'owner/repo' format to (knot, 'did/repo') by querying atproto repo records.
  def resolve_repo_identifier(owner_slash_repo: str) -> tuple[str, str]: """resolve owner/repo format to (knot, did/repo) for tangled XRPC Args: owner_slash_repo: repository identifier in "owner/repo" or "@owner/repo" format (e.g., "zzstoatzz.io/tangled-mcp" or "@zzstoatzz.io/tangled-mcp") Returns: tuple of (knot_url, repo_identifier) where: - knot_url: hostname of knot hosting the repo (e.g., "knot1.tangled.sh") - repo_identifier: "did/repo" format (e.g., "did:plc:.../tangled-mcp") Raises: ValueError: if format is invalid, handle cannot be resolved, or repo not found """ if "/" not in owner_slash_repo: raise ValueError( f"invalid repo format: '{owner_slash_repo}'. expected 'owner/repo'" ) owner, repo_name = owner_slash_repo.split("/", 1) client = _get_authenticated_client() # resolve owner (handle or DID) to DID if owner.startswith("did:"): owner_did = owner else: # strip @ prefix if present owner = owner.lstrip("@") # resolve handle to DID try: response = client.com.atproto.identity.resolve_handle( params={"handle": owner} ) owner_did = response.did except Exception as e: raise ValueError(f"failed to resolve handle '{owner}': {e}") from e # query owner's repo collection to find repo and get knot try: records = client.com.atproto.repo.list_records( models.ComAtprotoRepoListRecords.Params( repo=owner_did, collection="sh.tangled.repo", # correct collection name limit=100, ) ) except Exception as e: raise ValueError(f"failed to list repos for '{owner}': {e}") from e # find repo with matching name and extract knot for record in records.records: if (name := getattr(record.value, "name", None)) and name == repo_name: knot = getattr(record.value, "knot", None) if not knot: raise ValueError(f"repo '{repo_name}' has no knot information") return (knot, f"{owner_did}/{repo_name}") raise ValueError(f"repo '{repo_name}' not found for owner '{owner}'")