commit 0e7e83fe8b263772c3543a99b44a130e440b8800 Author: QiuSW <105186638@qq.com> Date: Fri Aug 7 22:18:16 2026 +0800 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..82d189e --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +# Local credentials +gitea.env + +# Python runtime artifacts +__pycache__/ +*.py[cod] + diff --git a/BACKUP_INFO.md b/BACKUP_INFO.md new file mode 100644 index 0000000..82b6623 --- /dev/null +++ b/BACKUP_INFO.md @@ -0,0 +1,24 @@ +# Gitea MCP 本机备份 + +备份日期：2026-08-07 + +## 内容 + +- `gitea-mcp.ps1`：Codex 当前实际使用的包装脚本，原路径为 `%USERPROFILE%\.codex\scripts\gitea-mcp.ps1`。 +- `gitea.env.example`：不含真实凭据的 Gitea 配置模板。 +- `gitea.env`：本机私有 Gitea 配置（被 Git 忽略），原路径为 `%USERPROFILE%\.codex\gitea.env`。 +- `package/gitea_mcp/`：本机维护的 `gitea-mcp 0.5.2.dev4` Python 运行包，包含 Gitea Wiki 读写工具，不含 `__pycache__`。 +- `package/gitea_mcp-0.5.2.dev4.dist-info/`：对应版本的 Wheel 安装元数据。 +- `reference/gitea-mcp-hardened.ps1`：`harness_coding_docs` 中固定 `0.5.1`、支持显式 HTTP 风险开关的通用加固版包装脚本。 + +## 恢复 + +1. 把 `gitea-mcp.ps1` 复制到 `%USERPROFILE%\.codex\scripts\gitea-mcp.ps1`。 +2. 参考 `gitea.env.example` 创建 `%USERPROFILE%\.codex\gitea.env`，填入实际地址和 Token；如有独立的安全备份，也可直接恢复该文件。 +3. 确认 Codex `config.toml` 的 Gitea MCP 命令仍指向上述包装脚本。 +4. 确认本机维护仓库及其虚拟环境位于 `D:\chengma\gitea-mcp`，或按实际位置修改包装脚本中的 `giteaMcpExecutable`。 +5. 通过包装脚本执行连接检查，并确认工具列表包含 6 个 Wiki 工具。 + +## 安全提示 + +`gitea.env` 含明文 Token，已通过 `.gitignore` 排除。不要强制提交、同步到公共云盘或发送给其他人；Token 轮换后应更新独立的安全备份。 diff --git a/gitea-mcp.ps1 b/gitea-mcp.ps1 new file mode 100644 index 0000000..c35c01c --- /dev/null +++ b/gitea-mcp.ps1 @@ -0,0 +1,57 @@ +$ErrorActionPreference = 'Stop' + +$envFile = Join-Path (Split-Path -Parent $PSScriptRoot) 'gitea.env' +if (-not (Test-Path -LiteralPath $envFile -PathType Leaf)) { + throw "Gitea MCP configuration file not found: $envFile" +} + +$values = @{} +foreach ($rawLine in Get-Content -LiteralPath $envFile) { + $line = $rawLine.Trim() + if (-not $line -or $line.StartsWith('#')) { + continue + } + + $parts = $line -split '=', 2 + if ($parts.Count -ne 2) { + throw "Invalid line in $envFile. Use KEY=VALUE format." + } + + $values[$parts[0].Trim()] = $parts[1].Trim() +} + +foreach ($name in @('GITEA_URL', 'GITEA_TOKEN')) { + if (-not $values.ContainsKey($name) -or [string]::IsNullOrWhiteSpace($values[$name])) { + throw "$name is missing or empty in $envFile" + } + Set-Item -Path "Env:$name" -Value $values[$name] +} + +# This self-hosted Gitea instance must not be sent through the system SOCKS proxy. +$noProxyHost = 'ilaer.eicp.net' +$existingNoProxy = @($env:NO_PROXY, $env:no_proxy) | + Where-Object { -not [string]::IsNullOrWhiteSpace($_) } | + Select-Object -First 1 +$noProxyEntries = @($existingNoProxy -split ',' | ForEach-Object { $_.Trim() } | Where-Object { $_ }) +if ($noProxyEntries -notcontains $noProxyHost) { + $noProxyEntries += $noProxyHost +} +$env:NO_PROXY = $noProxyEntries -join ',' +$env:no_proxy = $env:NO_PROXY + +# httpx initializes configured SOCKS transports before evaluating NO_PROXY. +# Clear proxy variables only for this MCP child process so it connects directly. +foreach ($proxyVariable in @('ALL_PROXY', 'all_proxy', 'HTTP_PROXY', 'http_proxy', 'HTTPS_PROXY', 'https_proxy')) { + Remove-Item -Path "Env:$proxyVariable" -ErrorAction SilentlyContinue +} + +# Run the locally maintained build so custom Wiki tools are not replaced by +# uvx resolving the latest published package on the next process start. +$giteaMcpExecutable = 'D:\chengma\gitea-mcp\.venv\Scripts\gitea-mcp.exe' +if (-not (Test-Path -LiteralPath $giteaMcpExecutable -PathType Leaf)) { + throw "Local Gitea MCP executable not found: $giteaMcpExecutable" +} + +# Keep stdout exclusively for MCP JSON-RPC. FastMCP diagnostics are noisy on stderr. +& $giteaMcpExecutable 2>> (Join-Path $PSScriptRoot ("gitea-mcp-$PID.stderr.log")) +exit $LASTEXITCODE diff --git a/gitea.env.example b/gitea.env.example new file mode 100644 index 0000000..1ff99d9 --- /dev/null +++ b/gitea.env.example @@ -0,0 +1,3 @@ +GITEA_URL=https://gitea.example.com +GITEA_TOKEN=your-personal-access-token + diff --git a/package/gitea_mcp-0.5.2.dev4.dist-info/METADATA b/package/gitea_mcp-0.5.2.dev4.dist-info/METADATA new file mode 100644 index 0000000..b6a87ad --- /dev/null +++ b/package/gitea_mcp-0.5.2.dev4.dist-info/METADATA @@ -0,0 +1,213 @@ +Metadata-Version: 2.4 +Name: gitea-mcp +Version: 0.5.2.dev4 +Summary: Model Context Protocol server for Gitea (and Forgejo, Codeberg). +Author-email: Sam Ware +License: MIT +Project-URL: Homepage, https://github.com/werebear73/gitea-mcp +Project-URL: Issues, https://github.com/werebear73/gitea-mcp/issues +Project-URL: Source, https://github.com/werebear73/gitea-mcp +Keywords: mcp,gitea,forgejo,codeberg,model-context-protocol,llm,ai +Classifier: Development Status :: 3 - Alpha +Classifier: Intended Audience :: Developers +Classifier: License :: OSI Approved :: MIT License +Classifier: Operating System :: OS Independent +Classifier: Programming Language :: Python :: 3 +Classifier: Programming Language :: Python :: 3.11 +Classifier: Programming Language :: Python :: 3.12 +Classifier: Programming Language :: Python :: 3.13 +Classifier: Topic :: Software Development :: Libraries :: Python Modules +Classifier: Topic :: Software Development :: Version Control :: Git +Requires-Python: >=3.11 +Description-Content-Type: text/markdown +License-File: LICENSE +Requires-Dist: fastmcp<3.0,>=2.0 +Requires-Dist: httpx>=0.27.0 +Requires-Dist: pydantic>=2.0.0 +Provides-Extra: dev +Requires-Dist: pytest>=8.0; extra == "dev" +Requires-Dist: pytest-asyncio>=0.23; extra == "dev" +Requires-Dist: pytest-httpx>=0.30; extra == "dev" +Requires-Dist: ruff>=0.5; extra == "dev" +Requires-Dist: mypy>=1.10; extra == "dev" +Requires-Dist: pre-commit>=3.5; extra == "dev" +Requires-Dist: build>=1.0; extra == "dev" +Requires-Dist: twine>=5.0; extra == "dev" +Dynamic: license-file + +# gitea-mcp + + + +A [Model Context Protocol](https://modelcontextprotocol.io) server for [Gitea](https://gitea.io) — lets AI assistants (Claude, ChatGPT, Copilot, and anything else that speaks MCP) read, create, and manage issues, repositories, and releases on any Gitea instance you can reach. + +Also works against **[Forgejo](https://forgejo.org)** and **[Codeberg](https://codeberg.org)** (API-compatible). + +## Why + +Self-hosted Gitea is a popular GitHub alternative for solo developers, small teams, and privacy-conscious organizations. With this MCP server installed, your AI assistant can: + +- File audit findings or refactor notes as Gitea issues without you leaving the chat +- Triage a repo's open issues in natural language +- Cut a release at the end of a coding session +- Comment on issues across multiple repos in one pass + +## Features + +| Resource | Tools | +| --- | --- | +| Issues | `create_issue`, `list_issues`, `get_issue`, `update_issue`, `add_comment` | +| Repos | `list_repos`, `list_labels`, `list_milestones`, `list_branches` | +| Pulls | `list_pull_requests`, `get_pull_request`, `add_comment_on_pr`, `create_pr`, `merge_pr` | +| Files | `read_file`, `commit_changes`, `create_branch` | +| Releases | `list_releases`, `create_release` | +| Wiki | `list_wiki_pages`, `get_wiki_page`, `list_wiki_revisions`, `create_wiki_page`, `update_wiki_page`, `delete_wiki_page` | +| Meta | `get_server_info`, `get_server_version` | + +- Bearer authentication via Personal Access Token (PAT) +- Async HTTP via `httpx` and `FastMCP` +- Works with self-hosted Gitea, Forgejo, and Codeberg + +## Quick Start + +### 1. Install + +```bash +pip install gitea-mcp +``` + +Or with [`uv`](https://docs.astral.sh/uv/): + +```bash +uv pip install gitea-mcp +``` + +### 2. Generate a Personal Access Token + +In your Gitea instance, go to **Settings → Applications → Generate New Token** and grant at least: + +- `read:repository` +- `write:issue` +- `read:user` + +Add `write:repository` if you also want to create releases or write wiki pages. + +### 3. Configure your MCP client + +**Claude Desktop (interactive):** run + +```bash +gitea-mcp init +``` + +It prompts for the Gitea URL and Personal Access Token, verifies the connection, and writes (or merges into) the right `claude_desktop_config.json` for your OS. Restart Claude Desktop and you're done. + +To check that the server can reach your Gitea instance at any time: + +```bash +gitea-mcp doctor +``` + +`doctor` reads `GITEA_URL` and `GITEA_TOKEN` from the environment, runs a `GET /api/v1/user`, and reports the authenticated username plus the state of the MCP tool surface. Exit `0` = ready; exit `1` = connection/load failure; exit `2` = missing config. + +**Any MCP client (manual):** add `gitea-mcp` to the client's MCP config. The recommended form uses `uvx` so the client launches the latest published wheel in an isolated env without needing `gitea-mcp` on its own PATH (this is what `gitea-mcp init` writes): + +```json +{ + "mcpServers": { + "gitea": { + "command": "uvx", + "args": ["gitea-mcp"], + "env": { + "GITEA_URL": "https://your-gitea-instance.example.com", + "GITEA_TOKEN": "your-personal-access-token" + } + } + } +} +``` + +If you'd rather use a globally pip-installed `gitea-mcp` binary, drop `args` and set `command` to `"gitea-mcp"` directly — works as long as the binary is on the MCP client's PATH at launch time. + +See [`mcp.json`](mcp.json) for a complete example. The same shape works for Claude Desktop, VS Code, Cowork, Claude Code, and any other MCP-compatible client. + +## Configuration + +Configuration is read from environment variables. + +| Variable | Required | Default | Description | +| --- | --- | --- | --- | +| `GITEA_URL` | Yes | — | Base URL of your Gitea instance (e.g., `https://gitea.example.com`) | +| `GITEA_TOKEN` | Yes | — | Personal Access Token from your Gitea user settings | +| `GITEA_TIMEOUT` | No | `30` | HTTP request timeout in seconds | +| `GITEA_MAX_RETRIES` | No | `3` | Max retries for transient failures on idempotent methods (`GET`/`PUT`/`DELETE`). Set to `0` to disable retries. `POST` and `PATCH` are never auto-retried — they could create duplicate issues, comments, or releases. `429 Too Many Requests` is retried for **any** method, honoring `Retry-After` when present. | +| `GITEA_RETRY_BASE_DELAY` | No | `0.5` | Base delay (seconds) for exponential backoff between retries. Effective delay grows as `base * 2^attempt` with jitter, capped at 4 seconds. | + +## Self-hosting / HTTP transport + +By default `gitea-mcp` runs in stdio mode — each MCP client (Claude Desktop, Cowork, etc.) launches its own subprocess on demand. For self-hosting one instance that multiple clients connect to over the network, use the streamable-HTTP transport: + +```bash +gitea-mcp serve --transport http --host 0.0.0.0 --port 8000 --path /mcp +``` + +All four flags can also be provided via environment variables (handy for Docker / systemd units): + +| Variable | Default | Flag | +| ------------------------- | ------------ | -------------- | +| `GITEA_MCP_TRANSPORT` | `stdio` | `--transport` | +| `GITEA_MCP_HOST` | `127.0.0.1` | `--host` | +| `GITEA_MCP_PORT` | `8000` | `--port` | +| `GITEA_MCP_PATH` | `/mcp` | `--path` | + +MCP clients connect to the resulting URL (e.g. `https://gitea-mcp.example.com/mcp`) just like they would to a local stdio server, except they share the one running instance. + +**Auth model (this release).** The server reads `GITEA_TOKEN` from its own environment, so any client that reaches the URL acts as that one Gitea user. Run it for yourself behind your own access controls (firewall, reverse-proxy auth, VPN, Tailscale). Multi-tenant bring-your-own-token is on the roadmap. + +The no-args invocation (`gitea-mcp` with no subcommand) still runs in stdio mode, so existing Claude Desktop / Cowork / Claude Code integrations are unaffected by this addition. + +## Compatibility + +| Server | Status | +| --- | --- | +| Gitea (self-hosted) | ✅ Primary target | +| Forgejo | ✅ Expected to work (API-compatible) | +| Codeberg | ✅ Expected to work (Codeberg runs Forgejo) | + +## Development + +```bash +git clone https://github.com/werebear73/gitea-mcp.git +cd gitea-mcp +pip install -e ".[dev]" +pre-commit install # commit-stage hooks (ruff + mypy) +pre-commit install --hook-type pre-push # push-stage hooks (pytest + build check) +pytest +``` + +The two-stage pre-commit policy keeps the commit loop snappy (lint + type only) while making `git push` block on the slow stuff that's actually caught CI/release bugs in the past — the full test suite and `python -m build && twine check dist/*`, which surfaces `setuptools_scm` version surprises before they reach a tag push. + +## Roadmap + +See [`docs/ROADMAP.md`](docs/ROADMAP.md) for what's shipped, what's next, and what's out of scope. + +## Publishing + +- MCP Registry metadata is tracked in [`server.json`](server.json). +- Smithery + MCP Registry publication steps are documented in [`docs/PUBLISHING.md`](docs/PUBLISHING.md). + +## Versioning + +Semantic versioning, derived from git tags via `setuptools_scm`. See [`VERSIONING.md`](VERSIONING.md) for the release process. + +## Contributing + +Issues and pull requests welcome. For substantial changes, please open an issue first to discuss the approach. + +## License + +[MIT](LICENSE) — use it however you like, including commercial products. + +--- + +Built by [Waretech Services](https://waretech.services). diff --git a/package/gitea_mcp-0.5.2.dev4.dist-info/RECORD b/package/gitea_mcp-0.5.2.dev4.dist-info/RECORD new file mode 100644 index 0000000..973458a --- /dev/null +++ b/package/gitea_mcp-0.5.2.dev4.dist-info/RECORD @@ -0,0 +1,23 @@ +gitea_mcp/__init__.py,sha256=9KYUxsefJXdcgx13lOdMKDTwtqW39J0oxAcNl4xM7n8,304 +gitea_mcp/_app.py,sha256=mNYrUKQi-o9DpkFJQLwUmM421Kj0EZBBdmOREFxifOU,1852 +gitea_mcp/_version.py,sha256=2JFIzAqJlZTk7JhAGeHGs4qJryKff5sFiNkbAeDnhEI,28 +gitea_mcp/client.py,sha256=eqSJFOFIPX6DlTVLzG0phHeAP41EorbfMMnGbDy0WL8,12123 +gitea_mcp/config.py,sha256=2QrZ7A5ZEI2QckWpc8HL8rJ5cAHx8k9L8QRD-24IlVw,2218 +gitea_mcp/doctor.py,sha256=5FGCyGwP6it-u3tsH0IV87T1lOzkg0BwGLefJgRmhOs,3707 +gitea_mcp/init.py,sha256=7X5vE84Q-H-CUE189zZXy3ru7BDUDWwM1_-URw0XjBQ,11373 +gitea_mcp/serve.py,sha256=0Un3iDYHnetwNKcXAQzABuCKtJRNqx2muEuZ7zToFmw,4920 +gitea_mcp/server.py,sha256=v63QSvmMyCefR9pjdYnrivGm-I1Tc1KZHsFHvUkjk0I,4711 +gitea_mcp/tools/__init__.py,sha256=XyQ5iUrz5alYHvWUH03LcGQ1dFJ_1aZ5lhof8XU8_K0,166 +gitea_mcp/tools/files.py,sha256=jV0RRlRCdNMi1DiOW2L2zp_i989bi3XTZCVger8rR-8,8612 +gitea_mcp/tools/issues.py,sha256=gVSFSzELoXh0-F8ceZGcvoLAI1vr21meFhowcI10APQ,9656 +gitea_mcp/tools/pulls.py,sha256=bZvySbFta7QiykMvVwOFWN71XFbKFKtwns2EpOHxgp4,7137 +gitea_mcp/tools/releases.py,sha256=i37PQ_AUz8zMmj4NvDK1eIbQQMqZUE-xvbA9fR0y8w0,3306 +gitea_mcp/tools/repos.py,sha256=od_xe3RHxgRN3bexRWUOW27OnQXIegIbGGXODt3GcN0,3423 +gitea_mcp/tools/server_info.py,sha256=PO8RmNwE7ENHcFon5ijvvZQV3pd-xjggXKLtWgrhAx0,3279 +gitea_mcp/tools/wiki.py,sha256=7h2CqK6E-FrJUGc1SG2QQbBxBY8hsMG2OBInWfP9EH0,9759 +gitea_mcp-0.5.2.dev4.dist-info/licenses/LICENSE,sha256=d23Qx0gy7rKXSx1QjoTq_i7-OBXtQNdQCzShSPgqiAI,1086 +gitea_mcp-0.5.2.dev4.dist-info/METADATA,sha256=wHZq5T9yVUCb9nLInn-kmOVr_NBHw6K1ozbSRW31XKs,9329 +gitea_mcp-0.5.2.dev4.dist-info/WHEEL,sha256=K260EYznzXsJYBQGqmI8VTxEdiZYNvDZwW9cBh9-_MA,91 +gitea_mcp-0.5.2.dev4.dist-info/entry_points.txt,sha256=PjRHlfQDMINFlBt4maFz_PWZ0p8t38Rw3br6-w-hZis,52 +gitea_mcp-0.5.2.dev4.dist-info/top_level.txt,sha256=anwGYTKslQDgerLVaH7Ded7ivajJbXfUAjcnmZeslWk,10 +gitea_mcp-0.5.2.dev4.dist-info/RECORD,, diff --git a/package/gitea_mcp-0.5.2.dev4.dist-info/WHEEL b/package/gitea_mcp-0.5.2.dev4.dist-info/WHEEL new file mode 100644 index 0000000..1d472b6 --- /dev/null +++ b/package/gitea_mcp-0.5.2.dev4.dist-info/WHEEL @@ -0,0 +1,5 @@ +Wheel-Version: 1.0 +Generator: setuptools (83.0.0) +Root-Is-Purelib: true +Tag: py3-none-any + diff --git a/package/gitea_mcp-0.5.2.dev4.dist-info/entry_points.txt b/package/gitea_mcp-0.5.2.dev4.dist-info/entry_points.txt new file mode 100644 index 0000000..4d17bdb --- /dev/null +++ b/package/gitea_mcp-0.5.2.dev4.dist-info/entry_points.txt @@ -0,0 +1,2 @@ +[console_scripts] +gitea-mcp = gitea_mcp.server:main diff --git a/package/gitea_mcp-0.5.2.dev4.dist-info/licenses/LICENSE b/package/gitea_mcp-0.5.2.dev4.dist-info/licenses/LICENSE new file mode 100644 index 0000000..1ef733a --- /dev/null +++ b/package/gitea_mcp-0.5.2.dev4.dist-info/licenses/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Sam Ware + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/package/gitea_mcp-0.5.2.dev4.dist-info/top_level.txt b/package/gitea_mcp-0.5.2.dev4.dist-info/top_level.txt new file mode 100644 index 0000000..99ee9aa --- /dev/null +++ b/package/gitea_mcp-0.5.2.dev4.dist-info/top_level.txt @@ -0,0 +1 @@ +gitea_mcp diff --git a/package/gitea_mcp/__init__.py b/package/gitea_mcp/__init__.py new file mode 100644 index 0000000..39e581e --- /dev/null +++ b/package/gitea_mcp/__init__.py @@ -0,0 +1,9 @@ +"""gitea-mcp — Model Context Protocol server for Gitea (and Forgejo, Codeberg).""" + +try: + from gitea_mcp._version import __version__ +except ImportError: + # Package not installed in editable mode or version file not yet generated + __version__ = "0.0.0.dev0" + +__all__ = ["__version__"] diff --git a/package/gitea_mcp/_app.py b/package/gitea_mcp/_app.py new file mode 100644 index 0000000..5e4283a --- /dev/null +++ b/package/gitea_mcp/_app.py @@ -0,0 +1,46 @@ +"""Internal seam holding the FastMCP singleton and client slot. + +This module exists so the ``FastMCP`` instance and the ``GiteaClient`` singleton +live in a module that is imported exactly once, regardless of how the package +entry point is launched. + +Why this matters: when a user runs ``python -m gitea_mcp.server``, Python loads +``server.py`` as ``__main__``. Any module that later imports +``gitea_mcp.server`` (e.g. one of the tool modules) causes Python to load +``server.py`` a *second* time, registered under its real name. If the +``FastMCP`` instance lived in ``server.py`` the tool decorators would register +against the second instance while the entry point's ``mcp.run()`` ran the +first — empty tools list, no errors. Keeping the singleton here means both +loads see the same ``mcp`` object. + +Tool modules should always import from here, not from ``gitea_mcp.server``. +""" + +from __future__ import annotations + +from fastmcp import FastMCP + +from gitea_mcp.client import GiteaClient + +# Module-level FastMCP instance. Tool modules register against this via +# ``@mcp.tool()`` at import time. +mcp: FastMCP = FastMCP("gitea-mcp") + +# Singleton client populated at startup by ``gitea_mcp.server.main()``. +# Tool modules access it via ``get_client()``. +_client: GiteaClient | None = None + + +def get_client() -> GiteaClient: + """Return the singleton :class:`GiteaClient`. + + Must be called after :func:`gitea_mcp.server.main` has initialized the + client. Tool functions call this at request time, not at import time. + """ + if _client is None: + raise RuntimeError( + "GiteaClient not initialized. The gitea-mcp server must be started " + "via the gitea-mcp entry point so the client is available before " + "any tools are called." + ) + return _client diff --git a/package/gitea_mcp/_version.py b/package/gitea_mcp/_version.py new file mode 100644 index 0000000..9954302 --- /dev/null +++ b/package/gitea_mcp/_version.py @@ -0,0 +1 @@ +__version__ = '0.5.2.dev4' diff --git a/package/gitea_mcp/client.py b/package/gitea_mcp/client.py new file mode 100644 index 0000000..26a2065 --- /dev/null +++ b/package/gitea_mcp/client.py @@ -0,0 +1,294 @@ +"""Async HTTP client wrapper for the Gitea REST API.""" + +from __future__ import annotations + +import asyncio +import random +from typing import Any + +import httpx + +# HTTP methods that are safe to retry on transient failures. +# POST and PATCH are excluded because retrying them could create duplicate +# issues / comments / releases — better to surface the failure to the caller. +_IDEMPOTENT_METHODS = frozenset({"GET", "PUT", "DELETE"}) + +# HTTP status codes that indicate transient server-side issues worth retrying +# (for idempotent methods only). +_RETRYABLE_STATUS_CODES = frozenset({502, 503, 504}) + +# Maximum backoff between retries, in seconds. +_MAX_BACKOFF_DELAY = 4.0 + + +class GiteaError(Exception): + """Base exception for Gitea client errors.""" + + +class GiteaAPIError(GiteaError): + """Raised when the Gitea API returns a non-success response.""" + + def __init__(self, status_code: int, message: str, method: str, url: str) -> None: + self.status_code = status_code + self.method = method + self.url = url + super().__init__(f"[{method} {url}] {status_code}: {message}") + + +class GiteaClient: + """Async HTTP client for the Gitea REST API. + + Uses Personal Access Token authentication via the + ``Authorization: token `` header (Gitea's convention; NOT Bearer). + One shared :class:`httpx.AsyncClient` per server lifetime. + + All paths passed to the verb methods are appended under ``/api/v1``; pass + ``/repos/{owner}/{repo}`` rather than the full URL. + + Two layers of verb methods: + + * **Untyped verbs** (``get``, ``post``, ``patch``, ``put``, ``delete``) + return the decoded JSON as :class:`typing.Any`. Use when the caller + doesn't care about the shape, or when the shape varies. + * **Typed verbs** (``get_json``, ``get_list``, ``post_json``, + ``patch_json``, ``put_json``, ``put_list``) wrap the untyped verbs + with a shape assertion and return ``dict[str, Any]`` or + ``list[dict[str, Any]]``. Use these in tool implementations so the + return type flows cleanly to the tool's annotated return without + needing a typed-intermediate variable (the pattern PR #4 had to + apply to every tool to satisfy strict mypy). + + **Retry policy** (applied by all verb methods): + + * Idempotent methods (``GET``, ``PUT``, ``DELETE``) are retried on + transient network errors (``httpx.ConnectError`` / + ``ReadTimeout`` / ``WriteTimeout``) and on 502 / 503 / 504 responses. + * ``POST`` and ``PATCH`` are **not** retried automatically — a retry + could create duplicate issues, comments, or releases. + * ``429 Too Many Requests`` is retried for **any** method, honoring the + ``Retry-After`` header if present; otherwise using the same + exponential backoff schedule. + * Backoff is exponential with jitter, capped at ``_MAX_BACKOFF_DELAY`` + seconds. Max attempts and base delay are constructor-configurable. + """ + + def __init__( + self, + base_url: str, + token: str, + timeout: float = 30.0, + max_retries: int = 3, + retry_base_delay: float = 0.5, + ) -> None: + self._base_url = base_url.rstrip("/") + self._max_retries = max_retries + self._retry_base_delay = retry_base_delay + self._client = httpx.AsyncClient( + base_url=self._base_url, + headers={ + "Authorization": f"token {token}", + "Accept": "application/json", + }, + timeout=timeout, + ) + + async def close(self) -> None: + """Close the underlying HTTP client. Safe to call multiple times.""" + await self._client.aclose() + + @property + def base_url(self) -> str: + """The Gitea instance base URL this client is configured against. + + Read-only — set at construction time. Useful for tools that report + server identity (e.g. ``get_server_info``). + """ + return self._base_url + + # ---- Untyped verbs (return Any) ---------------------------------------- + + async def get( + self, path: str, params: dict[str, Any] | None = None + ) -> Any: + response = await self._request_with_retry("GET", path, params=params) + return self._handle(response, method="GET", path=path) + + async def post(self, path: str, json: Any | None = None) -> Any: + response = await self._request_with_retry("POST", path, json=json) + return self._handle(response, method="POST", path=path) + + async def put(self, path: str, json: Any | None = None) -> Any: + response = await self._request_with_retry("PUT", path, json=json) + return self._handle(response, method="PUT", path=path) + + async def patch(self, path: str, json: Any | None = None) -> Any: + response = await self._request_with_retry("PATCH", path, json=json) + return self._handle(response, method="PATCH", path=path) + + async def delete(self, path: str) -> Any: + response = await self._request_with_retry("DELETE", path) + return self._handle(response, method="DELETE", path=path) + + # ---- Typed verbs (shape-asserted) -------------------------------------- + + async def get_json( + self, path: str, params: dict[str, Any] | None = None + ) -> dict[str, Any]: + """GET and decode as a JSON object. Raises :class:`GiteaError` on + non-object response.""" + return self._as_object(await self.get(path, params), method="GET", path=path) + + async def get_list( + self, path: str, params: dict[str, Any] | None = None + ) -> list[dict[str, Any]]: + """GET and decode as a JSON array of objects. Raises + :class:`GiteaError` on non-array response or non-object items.""" + return self._as_list(await self.get(path, params), method="GET", path=path) + + async def post_json(self, path: str, json: Any | None = None) -> dict[str, Any]: + """POST and decode the response as a JSON object.""" + return self._as_object(await self.post(path, json=json), method="POST", path=path) + + async def patch_json(self, path: str, json: Any | None = None) -> dict[str, Any]: + """PATCH and decode the response as a JSON object.""" + return self._as_object(await self.patch(path, json=json), method="PATCH", path=path) + + async def put_json(self, path: str, json: Any | None = None) -> dict[str, Any]: + """PUT and decode the response as a JSON object.""" + return self._as_object(await self.put(path, json=json), method="PUT", path=path) + + async def put_list( + self, path: str, json: Any | None = None + ) -> list[dict[str, Any]]: + """PUT and decode the response as a JSON array of objects. + + Specific to Gitea's ``PUT /repos/{owner}/{repo}/issues/{n}/labels``, + which returns the new label list (an array) rather than the issue. + """ + return self._as_list(await self.put(path, json=json), method="PUT", path=path) + + # ---- Retry & helpers --------------------------------------------------- + + async def _request_with_retry( + self, + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + json: Any | None = None, + ) -> httpx.Response: + """Issue a request, retrying transient failures per the class policy. + + Returns the final :class:`httpx.Response` (which may be a non-success + response — non-retryable errors and exhausted retries both return the + response so the caller's ``_handle`` can produce a proper + :class:`GiteaAPIError`). + """ + idempotent = method in _IDEMPOTENT_METHODS + url = self._api_path(path) + last_network_exc: Exception | None = None + + for attempt in range(self._max_retries + 1): + try: + response = await self._client.request( + method, url, params=params, json=json + ) + except (httpx.ConnectError, httpx.ReadTimeout, httpx.WriteTimeout) as exc: + last_network_exc = exc + if not idempotent or attempt >= self._max_retries: + raise + await asyncio.sleep(self._backoff_delay(attempt)) + continue + + # 429 honored for ANY method — the server is asking us to slow down. + if response.status_code == 429 and attempt < self._max_retries: + retry_after = self._parse_retry_after(response) + delay = retry_after if retry_after is not None else self._backoff_delay(attempt) + await asyncio.sleep(delay) + continue + + # Transient 5xx: retry only for idempotent methods. + if ( + response.status_code in _RETRYABLE_STATUS_CODES + and idempotent + and attempt < self._max_retries + ): + await asyncio.sleep(self._backoff_delay(attempt)) + continue + + return response + + # Loop body either returns a response or re-raises. Reaching here means + # the last iteration was a network exception that was already re-raised. + assert last_network_exc is not None + raise last_network_exc + + def _backoff_delay(self, attempt: int) -> float: + """Exponential backoff with jitter, capped at :data:`_MAX_BACKOFF_DELAY`.""" + base: float = min(_MAX_BACKOFF_DELAY, self._retry_base_delay * (2**attempt)) + jitter: float = random.uniform(0, base * 0.1) + result: float = base + jitter + return result + + @staticmethod + def _parse_retry_after(response: httpx.Response) -> float | None: + """Parse the ``Retry-After`` header value (seconds-int form only). + + Returns ``None`` if absent or in HTTP-date form (in which case the + caller falls back to its normal backoff schedule). + """ + raw = response.headers.get("retry-after") + if not raw: + return None + try: + return float(raw) + except ValueError: + return None + + @staticmethod + def _api_path(path: str) -> str: + """Prefix a relative path with /api/v1, leaving absolute API paths intact.""" + if path.startswith("/api/v1"): + return path + if path.startswith("/"): + return f"/api/v1{path}" + return f"/api/v1/{path}" + + def _handle(self, response: httpx.Response, method: str, path: str) -> Any: + if response.is_success: + if response.status_code == 204 or not response.content: + return None + return response.json() + message = response.text.strip() or response.reason_phrase + raise GiteaAPIError( + status_code=response.status_code, + message=message, + method=method, + url=str(response.request.url), + ) + + @staticmethod + def _as_object(raw: Any, *, method: str, path: str) -> dict[str, Any]: + if not isinstance(raw, dict): + raise GiteaError( + f"[{method} {path}] expected a JSON object response, got " + f"{type(raw).__name__}" + ) + result: dict[str, Any] = raw + return result + + @staticmethod + def _as_list(raw: Any, *, method: str, path: str) -> list[dict[str, Any]]: + if not isinstance(raw, list): + raise GiteaError( + f"[{method} {path}] expected a JSON array response, got " + f"{type(raw).__name__}" + ) + for i, item in enumerate(raw): + if not isinstance(item, dict): + raise GiteaError( + f"[{method} {path}] expected a JSON array of objects, " + f"item {i} is a {type(item).__name__}" + ) + result: list[dict[str, Any]] = raw + return result diff --git a/package/gitea_mcp/config.py b/package/gitea_mcp/config.py new file mode 100644 index 0000000..ed822b1 --- /dev/null +++ b/package/gitea_mcp/config.py @@ -0,0 +1,64 @@ +"""Configuration loaded from environment variables.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Config: + """Runtime configuration for gitea-mcp. + + Loaded once at startup from environment variables. Immutable thereafter. + """ + + base_url: str + token: str + timeout: float = 30.0 + max_retries: int = 3 + retry_base_delay: float = 0.5 + + @classmethod + def from_env(cls) -> Config: + """Load configuration from environment variables. + + Required: + GITEA_URL: Base URL of the Gitea instance (e.g. https://gitea.example.com) + GITEA_TOKEN: Personal Access Token + + Optional: + GITEA_TIMEOUT: HTTP request timeout in seconds (default: 30) + GITEA_MAX_RETRIES: Max retries for transient failures on idempotent + methods (GET/PUT/DELETE). Default 3. Set to 0 to disable retries. + GITEA_RETRY_BASE_DELAY: Base delay (seconds) for exponential backoff + between retries. Default 0.5. Effective delay is capped at 4s. + + Raises: + RuntimeError: if required variables are missing. + """ + base_url = os.environ.get("GITEA_URL", "").strip() + if not base_url: + raise RuntimeError( + "GITEA_URL environment variable is required. " + "Set it to the base URL of your Gitea instance." + ) + + token = os.environ.get("GITEA_TOKEN", "").strip() + if not token: + raise RuntimeError( + "GITEA_TOKEN environment variable is required. " + "Generate a Personal Access Token in Gitea: Settings -> Applications." + ) + + timeout = float(os.environ.get("GITEA_TIMEOUT", "30")) + max_retries = int(os.environ.get("GITEA_MAX_RETRIES", "3")) + retry_base_delay = float(os.environ.get("GITEA_RETRY_BASE_DELAY", "0.5")) + + return cls( + base_url=base_url.rstrip("/"), + token=token, + timeout=timeout, + max_retries=max_retries, + retry_base_delay=retry_base_delay, + ) diff --git a/package/gitea_mcp/doctor.py b/package/gitea_mcp/doctor.py new file mode 100644 index 0000000..5775797 --- /dev/null +++ b/package/gitea_mcp/doctor.py @@ -0,0 +1,104 @@ +"""Preflight check for gitea-mcp: verify GITEA_URL + PAT, report status. + +Run via ``gitea-mcp doctor``. Reads configuration from the same environment +variables the server uses (``GITEA_URL``, ``GITEA_TOKEN``, ``GITEA_TIMEOUT``) +and performs a ``GET /api/v1/user`` against the Gitea instance to confirm the +URL is reachable, the token works, and the response shape looks like a Gitea +user object. Then loads all tool modules to confirm the MCP surface is intact. + +Returns exit ``0`` on success, ``1`` on connection or load failure, ``2`` on +missing configuration. Useful before pointing Claude Desktop at the server, or +when an MCP client reports an empty tools list and you need to know whether +the problem is the connection or the integration. +""" + +from __future__ import annotations + +import argparse +import os +import sys + +from gitea_mcp import __version__ +from gitea_mcp.init import check_connection + + +def run(args: argparse.Namespace) -> int: + """Execute the doctor flow. Returns a process exit code.""" + url = (args.url or os.environ.get("GITEA_URL", "")).strip().rstrip("/") + if not url: + print( + "Error: GITEA_URL not set. Provide --url or set the GITEA_URL env var.", + file=sys.stderr, + ) + return 2 + + token = (args.token or os.environ.get("GITEA_TOKEN", "")).strip() + if not token: + print( + "Error: GITEA_TOKEN not set. Provide --token or set the GITEA_TOKEN env var.", + file=sys.stderr, + ) + return 2 + + timeout = float(os.environ.get("GITEA_TIMEOUT", "30")) + + print(f"gitea-mcp {__version__}") + print(f" Gitea URL : {url}") + print(f" Timeout : {timeout}s") + print() + + print(f"[1/2] Verifying connection to {url} ...") + try: + username = check_connection(url, token, timeout=timeout) + except RuntimeError as exc: + print(f" FAILED: {exc}", file=sys.stderr) + return 1 + print(f" OK — authenticated as '{username}'.") + + print("[2/2] Loading MCP tool modules ...") + try: + # Importing each tool module triggers its @mcp.tool() registrations + # against the FastMCP singleton in gitea_mcp._app. If any module fails + # to import, the server would also fail to start — catch it here so + # the user sees the real exception rather than a silent empty toolset. + from gitea_mcp.tools import issues, releases, repos, wiki # noqa: F401 + except Exception as exc: + print(f" FAILED: {exc}", file=sys.stderr) + return 1 + print(" OK — all tool modules loaded.") + + print() + print("All checks passed. gitea-mcp is ready to wire into your MCP client.") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + """argparse parser for the ``doctor`` subcommand.""" + parser = argparse.ArgumentParser( + prog="gitea-mcp doctor", + description=( + "Preflight check: verify the Gitea URL + Personal Access Token and " + "report status. Reads GITEA_URL and GITEA_TOKEN from the environment " + "by default." + ), + ) + parser.add_argument( + "--url", + help="Override the Gitea base URL (default: GITEA_URL env var).", + ) + parser.add_argument( + "--token", + help="Override the Personal Access Token (default: GITEA_TOKEN env var).", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + """Entry point for ``gitea-mcp doctor``.""" + parser = build_parser() + args = parser.parse_args(argv) + return run(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/package/gitea_mcp/init.py b/package/gitea_mcp/init.py new file mode 100644 index 0000000..2f8402f --- /dev/null +++ b/package/gitea_mcp/init.py @@ -0,0 +1,316 @@ +"""Interactive setup for adding gitea-mcp to Claude Desktop's config. + +Run via ``gitea-mcp init``. Prompts for the Gitea instance URL and Personal +Access Token, optionally verifies the connection, then merges a ``gitea`` +entry into the user's ``claude_desktop_config.json``. The existing file is +backed up with a timestamped ``.bak.*`` suffix before any write, and other +MCP servers in the same file are preserved. + +Non-interactive use: pass ``--url``, ``--token``, and ``--yes`` to skip the +prompts and confirmation (useful for CI or dotfile bootstrap scripts). +""" + +from __future__ import annotations + +import argparse +import getpass +import json +import os +import shutil +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import httpx + +# ---- Config path detection ------------------------------------------------- + + +def claude_desktop_config_path() -> Path: + """Return the OS-specific path to Claude Desktop's config file. + + Windows: ``%APPDATA%\\Claude\\claude_desktop_config.json`` + macOS: ``~/Library/Application Support/Claude/claude_desktop_config.json`` + Linux: ``~/.config/Claude/claude_desktop_config.json`` + + Neither the file nor its parent directory are guaranteed to exist; callers + must handle creation. The path is returned even on platforms where Claude + Desktop is not officially supported. + """ + if sys.platform.startswith("win"): + base = os.environ.get("APPDATA") or str(Path.home() / "AppData" / "Roaming") + return Path(base) / "Claude" / "claude_desktop_config.json" + if sys.platform == "darwin": + return ( + Path.home() + / "Library" + / "Application Support" + / "Claude" + / "claude_desktop_config.json" + ) + return Path.home() / ".config" / "Claude" / "claude_desktop_config.json" + + +# ---- Command auto-detection ------------------------------------------------ + + +def detect_command(prefer: str | None = None) -> tuple[str, list[str]]: + """Choose the ``(command, args)`` pair Claude Desktop should use to launch + gitea-mcp. + + Resolution order: + + 1. If ``prefer == "uvx"``: return ``("uvx", ["gitea-mcp"])`` unconditionally. + 2. If ``gitea-mcp`` is on PATH (the console script installed in a venv): + return its absolute path. This is the most reliable option on Windows, + where Claude Desktop's launch environment frequently does not match the + PATH the user sees in their shell. + 3. Fallback: ``(sys.executable, ["-m", "gitea_mcp.server"])``. Works + wherever the package itself is importable from the chosen Python. + """ + if prefer == "uvx": + return "uvx", ["gitea-mcp"] + on_path = shutil.which("gitea-mcp") + if on_path: + return on_path, [] + return sys.executable, ["-m", "gitea_mcp.server"] + + +# ---- Connection check ------------------------------------------------------ + + +def check_connection(url: str, token: str, timeout: float = 10.0) -> str: + """Verify the Gitea URL + PAT by calling ``GET /api/v1/user``. + + Returns the authenticated username on success. Raises ``RuntimeError`` with + a user-readable message on any failure (network, auth, malformed response). + """ + base = url.rstrip("/") + try: + with httpx.Client(timeout=timeout) as client: + response = client.get( + f"{base}/api/v1/user", + headers={ + "Authorization": f"token {token}", + "Accept": "application/json", + }, + ) + except httpx.RequestError as exc: + raise RuntimeError(f"Could not reach {base}: {exc}") from exc + if response.status_code == 401: + raise RuntimeError( + f"Authentication failed at {base}. Check the Personal Access Token." + ) + if not response.is_success: + message = response.text.strip() or response.reason_phrase + raise RuntimeError(f"{base} returned HTTP {response.status_code}: {message}") + payload: Any = response.json() + if not isinstance(payload, dict) or not isinstance(payload.get("login"), str): + raise RuntimeError( + f"{base} responded but the payload did not look like a Gitea user " + f"object (no 'login' field). Is the URL really a Gitea instance?" + ) + username: str = payload["login"] + return username + + +# ---- Config file manipulation ---------------------------------------------- + + +def load_config(path: Path) -> dict[str, Any]: + """Read an existing Claude Desktop config; return ``{}`` if missing or empty.""" + if not path.exists(): + return {} + text = path.read_text(encoding="utf-8").strip() + if not text: + return {} + loaded: Any = json.loads(text) + if not isinstance(loaded, dict): + raise RuntimeError( + f"{path} exists but does not contain a JSON object at the top level." + ) + result: dict[str, Any] = loaded + return result + + +def backup_config(path: Path) -> Path: + """Copy the existing config aside with a timestamped suffix. + + Returns the backup path. No-op (returns the original path unchanged) if the + source file does not exist. + """ + if not path.exists(): + return path + timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S") + backup = path.with_name(f"{path.name}.bak.{timestamp}") + shutil.copy2(path, backup) + return backup + + +def merge_server_entry( + config: dict[str, Any], + *, + name: str, + command: str, + args: list[str], + env: dict[str, str], +) -> dict[str, Any]: + """Insert or replace the named server entry under ``mcpServers``. + + Other servers under ``mcpServers`` are preserved. The input dict is mutated + in place and also returned for convenience. + """ + servers_obj: Any = config.setdefault("mcpServers", {}) + if not isinstance(servers_obj, dict): + raise RuntimeError( + "Existing claude_desktop_config.json has an 'mcpServers' key that " + "is not a JSON object. Refusing to overwrite — inspect and fix the " + "file manually." + ) + servers: dict[str, Any] = servers_obj + entry: dict[str, Any] = {"command": command, "env": env} + if args: + entry["args"] = args + servers[name] = entry + return config + + +def write_config(path: Path, config: dict[str, Any]) -> None: + """Write the config with two-space indent and a trailing newline.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8") + + +# ---- CLI ------------------------------------------------------------------- + + +def _prompt(label: str) -> str: + """Read a non-empty line from stdin, re-prompting on empty input.""" + while True: + raw = input(f"{label}: ").strip() + if raw: + return raw + print(" (value is required)") + + +def run(args: argparse.Namespace) -> int: + """Execute the init flow. Returns a process exit code.""" + url = (args.url or _prompt("Gitea base URL (e.g. https://gitea.example.com)")).rstrip("/") + if args.token: + token = args.token + else: + token = getpass.getpass("Personal Access Token (input hidden): ").strip() + if not token: + print("Error: no token provided.", file=sys.stderr) + return 2 + + config_path = Path(args.config_path) if args.config_path else claude_desktop_config_path() + server_name: str = args.name + prefer = None if args.command == "auto" else args.command + command, command_args = detect_command(prefer=prefer) + + if not args.skip_check: + print(f"Checking {url} ...") + try: + username = check_connection(url, token) + except RuntimeError as exc: + print(f" Connection check FAILED: {exc}", file=sys.stderr) + print(" Re-run with --skip-check to write the config anyway.", file=sys.stderr) + return 1 + print(f" OK — authenticated as '{username}'.") + + env = {"GITEA_URL": url, "GITEA_TOKEN": token} + + print() + print(f"Will write the following entry to {config_path}:") + print(f" mcpServers.{server_name}.command = {command}") + if command_args: + print(f" mcpServers.{server_name}.args = {command_args}") + print(f" mcpServers.{server_name}.env = {{GITEA_URL=..., GITEA_TOKEN=***}}") + if not args.yes: + confirm = input("Proceed? [y/N]: ").strip().lower() + if confirm not in {"y", "yes"}: + print("Aborted.") + return 0 + + try: + config = load_config(config_path) + except (json.JSONDecodeError, RuntimeError) as exc: + print(f"Error reading {config_path}: {exc}", file=sys.stderr) + return 1 + + backup = backup_config(config_path) + merge_server_entry( + config, + name=server_name, + command=command, + args=command_args, + env=env, + ) + write_config(config_path, config) + + print() + if backup != config_path: + print(f"Backed up existing config to {backup}.") + print(f"Wrote {config_path}.") + print( + "Note: GITEA_TOKEN is stored in plaintext at the path above. " + "Restrict file permissions if this is a shared machine." + ) + print("Restart Claude Desktop to load the new server.") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + """argparse parser for the ``init`` subcommand.""" + parser = argparse.ArgumentParser( + prog="gitea-mcp init", + description=( + "Add gitea-mcp to Claude Desktop's claude_desktop_config.json. " + "Prompts interactively unless --url, --token, and --yes are provided." + ), + ) + parser.add_argument("--url", help="Gitea base URL (e.g. https://gitea.example.com)") + parser.add_argument("--token", help="Personal Access Token") + parser.add_argument( + "--name", + default="gitea", + help="Server key under mcpServers (default: gitea). Use a unique value " + "if you run multiple Gitea instances.", + ) + parser.add_argument( + "--command", + choices=["auto", "uvx"], + default="auto", + help="Launch command to write. 'auto' (default) uses the gitea-mcp " + "console script if on PATH, else 'python -m gitea_mcp.server'. " + "'uvx' writes 'uvx gitea-mcp' (requires gitea-mcp on PyPI).", + ) + parser.add_argument( + "--config-path", + help="Override the Claude Desktop config path (default: OS-specific).", + ) + parser.add_argument( + "--skip-check", + action="store_true", + help="Skip the GET /api/v1/user connection check before writing.", + ) + parser.add_argument( + "--yes", + "-y", + action="store_true", + help="Skip the final confirmation prompt.", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + """Entry point for ``gitea-mcp init``.""" + parser = build_parser() + args = parser.parse_args(argv) + return run(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/package/gitea_mcp/serve.py b/package/gitea_mcp/serve.py new file mode 100644 index 0000000..cda5703 --- /dev/null +++ b/package/gitea_mcp/serve.py @@ -0,0 +1,142 @@ +"""Transport-selectable runner for the gitea-mcp MCP server. + +Adds a ``gitea-mcp serve`` subcommand that lets the operator pick between +stdio (the default — what Claude Desktop and other local MCP clients use) +and HTTP transport (for self-hosting one running instance that multiple +clients connect to over the network). + +The no-args invocation ``gitea-mcp`` continues to run in stdio mode via the +dispatcher in :mod:`gitea_mcp.server` so existing Claude Desktop / Cowork / +Claude Code integrations and the dual-load regression test are unaffected. + +**Auth model for HTTP transport (v0.5.0).** Single-user — the server reads +``GITEA_TOKEN`` from its own environment exactly as the stdio mode does, and +any client that reaches the URL acts as that one user against Gitea. This is +appropriate for self-hosted personal use behind your own access controls +(firewall, reverse-proxy auth, VPN). Multi-tenant bring-your-own-token is a +real auth-integration project deferred to a later release. +""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import os +import sys + +from gitea_mcp import _app +from gitea_mcp._app import mcp +from gitea_mcp.client import GiteaClient +from gitea_mcp.config import Config + +# Tool modules must be imported so their @mcp.tool() decorators fire and +# register against the singleton in _app. Importing them here (in addition +# to in server.py) keeps `python -m gitea_mcp.serve` viable as a direct +# invocation path, though the canonical entry is `gitea-mcp serve`. +from gitea_mcp.tools import ( # noqa: F401 + files, + issues, + pulls, + releases, + repos, + server_info, + wiki, +) + +_DEFAULT_HOST = "127.0.0.1" +_DEFAULT_PORT = 8000 +_DEFAULT_PATH = "/mcp" + + +def run(args: argparse.Namespace) -> int: + """Execute the serve flow. Returns a process exit code.""" + config = Config.from_env() + client = GiteaClient( + base_url=config.base_url, + token=config.token, + timeout=config.timeout, + max_retries=config.max_retries, + retry_base_delay=config.retry_base_delay, + ) + _app._client = client + + try: + if args.transport == "stdio": + mcp.run() + else: + # FastMCP accepts "http" / "streamable-http" / "sse" — we expose + # the friendlier "http" alias on the CLI which FastMCP maps to + # streamable-http internally. + mcp.run( + transport="http", + host=args.host, + port=args.port, + path=args.path, + ) + finally: + # Best-effort cleanup. If the event loop is already closed, ignore. + with contextlib.suppress(RuntimeError): + asyncio.run(client.close()) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + """argparse parser for the ``serve`` subcommand.""" + parser = argparse.ArgumentParser( + prog="gitea-mcp serve", + description=( + "Start the gitea-mcp MCP server with the chosen transport. " + "Defaults match the no-args `gitea-mcp` invocation (stdio) so " + "existing Claude Desktop / Cowork integrations are unaffected; " + "pass --transport http to self-host one instance for multiple clients." + ), + ) + parser.add_argument( + "--transport", + choices=["stdio", "http"], + default=os.environ.get("GITEA_MCP_TRANSPORT", "stdio"), + help=( + "Transport to run. 'stdio' (default) for local MCP-client launch; " + "'http' for self-hosted streamable-HTTP. " + "Env: GITEA_MCP_TRANSPORT." + ), + ) + parser.add_argument( + "--host", + default=os.environ.get("GITEA_MCP_HOST", _DEFAULT_HOST), + help=( + f"Bind address for HTTP transport (default: {_DEFAULT_HOST}). " + "Use 0.0.0.0 to listen on all interfaces (Docker, public hosting). " + "Env: GITEA_MCP_HOST. Ignored for stdio." + ), + ) + parser.add_argument( + "--port", + type=int, + default=int(os.environ.get("GITEA_MCP_PORT", _DEFAULT_PORT)), + help=( + f"TCP port for HTTP transport (default: {_DEFAULT_PORT}). " + "Env: GITEA_MCP_PORT. Ignored for stdio." + ), + ) + parser.add_argument( + "--path", + default=os.environ.get("GITEA_MCP_PATH", _DEFAULT_PATH), + help=( + f"URL path the MCP endpoint serves at (default: {_DEFAULT_PATH}). " + "Env: GITEA_MCP_PATH. Ignored for stdio." + ), + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + """Entry point for ``gitea-mcp serve``.""" + parser = build_parser() + args = parser.parse_args(argv) + return run(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/package/gitea_mcp/server.py b/package/gitea_mcp/server.py new file mode 100644 index 0000000..dcac4a0 --- /dev/null +++ b/package/gitea_mcp/server.py @@ -0,0 +1,132 @@ +"""MCP server entry point. Initializes the client and runs stdio transport. + +The ``FastMCP`` singleton lives in :mod:`gitea_mcp._app`, not here. This module +is intentionally a thin launcher so that ``python -m gitea_mcp.server`` is safe +to load twice (once as ``__main__``, once under its real name when a tool +module indirectly imports it). See :mod:`gitea_mcp._app` for the full rationale. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import sys + +from gitea_mcp import __version__, _app +from gitea_mcp._app import mcp +from gitea_mcp.client import GiteaClient +from gitea_mcp.config import Config + +# Import tool modules so their @mcp.tool() registrations execute on module +# load. Ordering doesn't matter; each module registers against the shared +# ``mcp`` instance in ``_app``. +from gitea_mcp.tools import ( # noqa: E402, F401 + files, + issues, + pulls, + releases, + repos, + server_info, + wiki, +) + + +def main() -> None: + """Console-script entry point. + + Dispatches based on the first positional argument: + + * No arguments (the default Claude Desktop / Claude Code invocation): + runs the MCP server over stdio. This path is byte-identical to the + pre-v0.1.2 behavior and is guarded by ``tests/test_subprocess_launch``. + * ``--help`` / ``-h``: prints top-level help. + * ``--version`` / ``-V``: prints ``gitea-mcp ``. + * ``init``: hands off to :func:`gitea_mcp.init.main` for interactive setup. + * ``doctor``: hands off to :func:`gitea_mcp.doctor.main` for the preflight. + * Anything else: prints an unknown-subcommand error and exits 2. + + Subcommand-level flag parsing (e.g. ``gitea-mcp init --help``) lives in + the subcommand modules' own argparse parsers; this dispatcher passes the + remaining argv through unchanged. + """ + argv = sys.argv[1:] + + if not argv: + _run_server() + return + + first = argv[0] + + if first in ("-h", "--help"): + _print_help() + return + + if first in ("-V", "--version"): + print(f"gitea-mcp {__version__}") + return + + if first == "init": + from gitea_mcp.init import main as init_main + + sys.exit(init_main(argv[1:])) + + if first == "doctor": + from gitea_mcp.doctor import main as doctor_main + + sys.exit(doctor_main(argv[1:])) + + if first == "serve": + from gitea_mcp.serve import main as serve_main + + sys.exit(serve_main(argv[1:])) + + print(f"Error: unknown subcommand '{first}'.", file=sys.stderr) + print("Run 'gitea-mcp --help' for usage.", file=sys.stderr) + sys.exit(2) + + +def _print_help() -> None: + """Print the top-level help text.""" + print( + "gitea-mcp — Model Context Protocol server for Gitea (and Forgejo, Codeberg)\n" + "\n" + "Usage:\n" + " gitea-mcp Start the MCP server (stdio transport).\n" + " This is what Claude Desktop / Claude Code\n" + " invoke. Requires GITEA_URL and GITEA_TOKEN\n" + " environment variables.\n" + " gitea-mcp serve [opts] Start the server with a chosen transport\n" + " (stdio or HTTP). Use --transport http for\n" + " self-hosting one instance for multiple clients.\n" + " Run 'gitea-mcp serve --help' for options.\n" + " gitea-mcp init [opts] Interactive setup: add gitea-mcp to\n" + " claude_desktop_config.json.\n" + " Run 'gitea-mcp init --help' for options.\n" + " gitea-mcp doctor [opts] Preflight check: verify GITEA_URL + token\n" + " and report status.\n" + " Run 'gitea-mcp doctor --help' for options.\n" + " gitea-mcp --version, -V Print version and exit.\n" + " gitea-mcp --help, -h This message.\n" + ) + + +def _run_server() -> None: + config = Config.from_env() + client = GiteaClient( + base_url=config.base_url, + token=config.token, + timeout=config.timeout, + max_retries=config.max_retries, + retry_base_delay=config.retry_base_delay, + ) + _app._client = client + try: + mcp.run() + finally: + # Best-effort cleanup. If the event loop is already closed, ignore. + with contextlib.suppress(RuntimeError): + asyncio.run(client.close()) + + +if __name__ == "__main__": + main() diff --git a/package/gitea_mcp/tools/__init__.py b/package/gitea_mcp/tools/__init__.py new file mode 100644 index 0000000..027a0fd --- /dev/null +++ b/package/gitea_mcp/tools/__init__.py @@ -0,0 +1,5 @@ +"""MCP tool definitions, organized by Gitea resource family. + +Each submodule registers its tools against the FastMCP instance defined in +``gitea_mcp._app``. +""" diff --git a/package/gitea_mcp/tools/files.py b/package/gitea_mcp/tools/files.py new file mode 100644 index 0000000..4c43fd6 --- /dev/null +++ b/package/gitea_mcp/tools/files.py @@ -0,0 +1,245 @@ +"""MCP tools for Gitea file operations and pull-request creation. + +This is the surface that lets the LLM actually *change code* via gitea-mcp, +not just read and discuss it. Four tools form a complete edit workflow: + +1. :func:`read_file` — read a file's current content (optionally pinned to a + specific branch or commit). +2. :func:`create_branch` — open a feature branch off the default branch (or + any specified base). +3. :func:`commit_changes` — write a single file's new content on a branch; + automatically detects whether the file is new or existing (GET to fetch + the current SHA; 404 → POST create; existing → PUT update). +4. :func:`create_pr` — open a pull request from the feature branch back to + the base. + +Multi-file commits via the Git Trees API are deliberately out of scope for +this MVP — single-file `commit_changes` is the simpler, safer pattern. +""" + +from __future__ import annotations + +import base64 +from typing import Annotated, Any + +from mcp.types import ToolAnnotations +from pydantic import Field + +from gitea_mcp._app import get_client, mcp +from gitea_mcp.client import GiteaAPIError + +_READ_ONLY = ToolAnnotations(readOnlyHint=True, openWorldHint=True) + + +@mcp.tool(annotations=_READ_ONLY) +async def read_file( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + path: Annotated[ + str, + Field(description="Path within the repository, e.g. 'src/gitea_mcp/server.py'"), + ], + ref: Annotated[ + str | None, + Field( + description=( + "Branch name, tag name, or commit SHA to read from. " + "Omit to read from the repository's default branch." + ), + ), + ] = None, +) -> dict[str, Any]: + """Read a file's content from a Gitea repository. + + Returns Gitea's ``ContentsResponse`` shape extended with a ``text`` field + containing the decoded file content as a UTF-8 string (or ``None`` if the + content isn't valid UTF-8 — binary file). The raw base64 ``content`` and + Gitea's ``encoding`` are preserved so callers can re-decode if needed. + + The ``sha`` field in the response is what :func:`commit_changes` would + need to update this file — but ``commit_changes`` fetches it internally, + so callers don't usually need to pass it forward. + """ + params: dict[str, Any] = {} + if ref is not None: + params["ref"] = ref + response = await get_client().get_json( + f"/repos/{owner}/{repo}/contents/{path}", + params=params or None, + ) + text: str | None = None + encoding = response.get("encoding") + content = response.get("content") + if encoding == "base64" and isinstance(content, str): + try: + text = base64.b64decode(content).decode("utf-8") + except (ValueError, UnicodeDecodeError): + text = None + response["text"] = text + return response + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + openWorldHint=True, + ) +) +async def create_branch( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + new_branch_name: Annotated[ + str, + Field(description="Name for the new branch (must not already exist)"), + ], + old_branch_name: Annotated[ + str | None, + Field( + description=( + "Existing branch to fork from. Omit to use the repository's " + "default branch." + ), + ), + ] = None, +) -> dict[str, Any]: + """Create a new branch in a repository. + + Returns the created Gitea Branch object. Fails cleanly via + :class:`GiteaAPIError` if the new branch name already exists. + """ + payload: dict[str, Any] = {"new_branch_name": new_branch_name} + if old_branch_name is not None: + payload["old_branch_name"] = old_branch_name + return await get_client().post_json( + f"/repos/{owner}/{repo}/branches", + json=payload, + ) + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + # Overwrites existing file content (when updating). Reversible via git, + # but a user-visible state change worth gating on confirmation. + destructiveHint=True, + openWorldHint=True, + ) +) +async def commit_changes( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + branch: Annotated[ + str, + Field(description="Branch to commit to (must exist; use create_branch first)"), + ], + path: Annotated[ + str, + Field(description="File path within the repository, e.g. 'README.md'"), + ], + content: Annotated[ + str, + Field(description="New file content (UTF-8 text; binary files not supported)"), + ], + message: Annotated[str, Field(description="Commit message")], +) -> dict[str, Any]: + """Create or update a single file on a branch in one commit. + + Auto-detects whether the file exists: + + - **File does not exist on the branch:** Gitea returns 404 to the SHA + lookup; we ``POST`` to create the file. + - **File exists:** we use its current SHA and ``PUT`` to update it. + + Returns Gitea's ``FileResponse`` shape (the resulting commit + content + metadata). Raises :class:`GiteaAPIError` on conflicts (e.g. concurrent + update changed the SHA between our lookup and our write — caller should + re-read and retry). + + Single-file only. Multi-file commits would require Gitea's Git Trees + API and are deliberately out of scope for this tool. + """ + client = get_client() + encoded_content = base64.b64encode(content.encode("utf-8")).decode("ascii") + payload: dict[str, Any] = { + "content": encoded_content, + "message": message, + "branch": branch, + } + + # SHA lookup — determines create vs update. + existing_sha: str | None = None + try: + existing = await client.get_json( + f"/repos/{owner}/{repo}/contents/{path}", + params={"ref": branch}, + ) + sha_value = existing.get("sha") + if isinstance(sha_value, str): + existing_sha = sha_value + except GiteaAPIError as exc: + if exc.status_code != 404: + raise + # 404 = file doesn't exist; fall through to create. + + if existing_sha is not None: + payload["sha"] = existing_sha + return await client.put_json( + f"/repos/{owner}/{repo}/contents/{path}", + json=payload, + ) + return await client.post_json( + f"/repos/{owner}/{repo}/contents/{path}", + json=payload, + ) + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + openWorldHint=True, + ) +) +async def create_pr( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + head: Annotated[ + str, + Field( + description=( + "Source branch (the branch containing your changes). " + "Same-repo only; cross-fork PRs not supported by this tool." + ), + ), + ], + base: Annotated[ + str, + Field(description="Target branch (where the PR should merge into, e.g. 'main')"), + ], + title: Annotated[str, Field(description="Pull request title")], + body: Annotated[str, Field(description="PR description in Markdown")] = "", + draft: Annotated[ + bool, + Field(description="Open as a draft PR (cannot be merged until marked ready)"), + ] = False, +) -> dict[str, Any]: + """Open a new pull request from ``head`` into ``base``. + + Returns the created Gitea PullRequest object (including the assigned + number, URL, and merge status). Raises :class:`GiteaAPIError` if the + head branch doesn't exist, there are no commits between head and base, + or an open PR already exists for this branch pair. + """ + payload: dict[str, Any] = { + "head": head, + "base": base, + "title": title, + "body": body, + } + if draft: + payload["draft"] = True + return await get_client().post_json( + f"/repos/{owner}/{repo}/pulls", + json=payload, + ) diff --git a/package/gitea_mcp/tools/issues.py b/package/gitea_mcp/tools/issues.py new file mode 100644 index 0000000..f95ab81 --- /dev/null +++ b/package/gitea_mcp/tools/issues.py @@ -0,0 +1,255 @@ +"""MCP tools for Gitea issues.""" + +from __future__ import annotations + +from typing import Annotated, Any + +from mcp.types import ToolAnnotations +from pydantic import Field + +from gitea_mcp._app import get_client, mcp +from gitea_mcp.client import GiteaClient, GiteaError + +# ---- Internal helpers ------------------------------------------------------ + + +async def _list_all_labels( + client: GiteaClient, owner: str, repo: str +) -> list[dict[str, Any]]: + """Page through every label defined in a repository.""" + all_labels: list[dict[str, Any]] = [] + page = 1 + while True: + batch = await client.get_list( + f"/repos/{owner}/{repo}/labels", + params={"page": page, "limit": 50}, + ) + if not batch: + break + all_labels.extend(batch) + if len(batch) < 50: + break + page += 1 + return all_labels + + +async def _resolve_label_ids( + client: GiteaClient, owner: str, repo: str, label_names: list[str] +) -> list[int]: + """Resolve a list of label names to the integer IDs Gitea's issue API expects. + + Gitea's create-issue and replace-issue-labels endpoints take ``labels`` as a + list of integer IDs, not names. This helper fetches the repo's labels once + and maps the names in. Raises :class:`GiteaError` if any name doesn't match + a defined label, including the available label names in the error message + so the caller knows what's valid. + """ + if not label_names: + return [] + all_labels = await _list_all_labels(client, owner, repo) + name_to_id = {label["name"]: label["id"] for label in all_labels} + missing = [name for name in label_names if name not in name_to_id] + if missing: + raise GiteaError( + f"Labels not found in {owner}/{repo}: {missing}. " + f"Available labels: {sorted(name_to_id.keys())}" + ) + return [name_to_id[name] for name in label_names] + + +# ---- Tools ----------------------------------------------------------------- + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + openWorldHint=True, + ) +) +async def create_issue( + owner: Annotated[str, Field(description="Repository owner (user or organization name)")], + repo: Annotated[str, Field(description="Repository name")], + title: Annotated[str, Field(description="Issue title")], + body: Annotated[str, Field(description="Issue body in Markdown")] = "", + labels: Annotated[ + list[str] | None, + Field(description="Label names to apply to the new issue (resolved to IDs automatically)"), + ] = None, + assignees: Annotated[ + list[str] | None, + Field(description="Usernames to assign to the new issue"), + ] = None, + milestone: Annotated[int | None, Field(description="Milestone ID to attach")] = None, +) -> dict[str, Any]: + """Create a new issue in a Gitea repository. + + Returns the full Gitea Issue object including the assigned number, URL, and + metadata. Label names are resolved to IDs against the repository's label + set; an unknown label name fails the call cleanly with the list of valid + names. + """ + client = get_client() + payload: dict[str, Any] = {"title": title, "body": body} + if assignees: + payload["assignees"] = assignees + if milestone is not None: + payload["milestone"] = milestone + if labels: + payload["labels"] = await _resolve_label_ids(client, owner, repo, labels) + return await client.post_json(f"/repos/{owner}/{repo}/issues", json=payload) + + +@mcp.tool( + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True) +) +async def list_issues( + owner: Annotated[str, Field(description="Repository owner (user or organization name)")], + repo: Annotated[str, Field(description="Repository name")], + state: Annotated[ + str, Field(description="Filter by state: 'open', 'closed', or 'all'") + ] = "open", + labels: Annotated[ + str | None, + Field(description="Comma-separated label names to filter by"), + ] = None, + assignee: Annotated[ + str | None, + Field(description="Filter to issues assigned to this username"), + ] = None, + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, + limit: Annotated[int, Field(description="Items per page (max 50)")] = 30, +) -> list[dict[str, Any]]: + """List issues in a Gitea repository. + + Pull requests are excluded; only true issues are returned. Filters compose + (state AND labels AND assignee). + """ + client = get_client() + params: dict[str, Any] = { + "state": state, + "type": "issues", # exclude pull requests + "page": page, + "limit": limit, + } + if labels: + params["labels"] = labels + if assignee: + params["assigned_by"] = assignee + return await client.get_list(f"/repos/{owner}/{repo}/issues", params=params) + + +@mcp.tool( + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True) +) +async def get_issue( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + issue_number: Annotated[int, Field(description="Issue number (the #N in the URL)")], +) -> dict[str, Any]: + """Get a single issue by number, including all of its comments. + + The returned object is the standard Gitea Issue payload, with an additional + ``comments_list`` field containing the full list of Comment objects. The + existing top-level ``comments`` integer field (comment count) is preserved. + """ + client = get_client() + issue = await client.get_json(f"/repos/{owner}/{repo}/issues/{issue_number}") + issue["comments_list"] = await client.get_list( + f"/repos/{owner}/{repo}/issues/{issue_number}/comments" + ) + return issue + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + # Can close the issue and clear labels — both reversible but + # user-visible side effects, so clients should gate on confirmation + # rather than auto-approve. + destructiveHint=True, + idempotentHint=True, + openWorldHint=True, + ) +) +async def update_issue( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + issue_number: Annotated[int, Field(description="Issue number")], + title: Annotated[str | None, Field(description="New title")] = None, + body: Annotated[str | None, Field(description="New body (Markdown)")] = None, + state: Annotated[ + str | None, + Field(description="New state: 'open' or 'closed'"), + ] = None, + labels: Annotated[ + list[str] | None, + Field(description="Replace labels with this exact set (names; pass [] to clear)"), + ] = None, + assignees: Annotated[ + list[str] | None, + Field(description="Replace assignees with this exact set of usernames"), + ] = None, + milestone: Annotated[ + int | None, + Field(description="Milestone ID to attach; pass 0 to clear the milestone"), + ] = None, +) -> dict[str, Any]: + """Update an existing issue's title, body, state, assignees, milestone, or labels. + + Each argument is independent — pass only the fields you want to change. + Labels are replaced atomically against the new set (passing ``[]`` removes + all labels). Other list fields (assignees) follow the same replace semantics. + """ + client = get_client() + payload: dict[str, Any] = {} + if title is not None: + payload["title"] = title + if body is not None: + payload["body"] = body + if state is not None: + payload["state"] = state + if assignees is not None: + payload["assignees"] = assignees + if milestone is not None: + # Gitea convention: milestone=0 in the request clears the milestone. + # We send null in that case, which Gitea also accepts and is unambiguous. + payload["milestone"] = milestone if milestone > 0 else None + + if payload: + issue = await client.patch_json( + f"/repos/{owner}/{repo}/issues/{issue_number}", json=payload + ) + else: + # No PATCH-level changes — fetch the current issue so the caller still + # gets the up-to-date object after the labels update below. + issue = await client.get_json(f"/repos/{owner}/{repo}/issues/{issue_number}") + + if labels is not None: + label_ids = await _resolve_label_ids(client, owner, repo, labels) + issue["labels"] = await client.put_list( + f"/repos/{owner}/{repo}/issues/{issue_number}/labels", + json={"labels": label_ids}, + ) + + return issue + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + openWorldHint=True, + ) +) +async def add_comment( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + issue_number: Annotated[int, Field(description="Issue number")], + body: Annotated[str, Field(description="Comment body (Markdown)")], +) -> dict[str, Any]: + """Add a comment to an existing issue. Returns the created Comment object.""" + return await get_client().post_json( + f"/repos/{owner}/{repo}/issues/{issue_number}/comments", + json={"body": body}, + ) diff --git a/package/gitea_mcp/tools/pulls.py b/package/gitea_mcp/tools/pulls.py new file mode 100644 index 0000000..d8af6f3 --- /dev/null +++ b/package/gitea_mcp/tools/pulls.py @@ -0,0 +1,188 @@ +"""MCP tools for Gitea branches and pull requests. + +Grouped together because the workflow is shared: inspecting branches, listing +PRs against them, reading a PR's discussion thread, and adding a comment to +move the conversation forward. + +A note on the Gitea API: pull requests and issues share the same number +namespace and the same comments endpoint (``/repos/{owner}/{repo}/issues/{n}/comments``). +That's why ``add_comment_on_pr`` posts to ``/issues/{pull_number}/comments`` +rather than a hypothetical ``/pulls/{pull_number}/comments`` — Gitea simply +doesn't have one for non-review comments. Inline review comments (those +attached to specific diff lines) live under ``/pulls/{n}/reviews`` and are +out of scope for this MVP. +""" + +from __future__ import annotations + +from typing import Annotated, Any + +from mcp.types import ToolAnnotations +from pydantic import Field + +from gitea_mcp._app import get_client, mcp + +_READ_ONLY = ToolAnnotations(readOnlyHint=True, openWorldHint=True) + + +@mcp.tool(annotations=_READ_ONLY) +async def list_branches( + owner: Annotated[str, Field(description="Repository owner (user or organization name)")], + repo: Annotated[str, Field(description="Repository name")], + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, + limit: Annotated[int, Field(description="Items per page (max 50)")] = 30, +) -> list[dict[str, Any]]: + """List branches in a repository. + + Returns the standard Gitea Branch object array: ``[{name, commit: {id, ...}, + protected, ...}]``. Useful for inspecting available targets before creating + a release, opening a PR, or filing a fix against a specific branch. + """ + return await get_client().get_list( + f"/repos/{owner}/{repo}/branches", + params={"page": page, "limit": limit}, + ) + + +@mcp.tool(annotations=_READ_ONLY) +async def list_pull_requests( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + state: Annotated[ + str, Field(description="Filter by state: 'open', 'closed', or 'all'") + ] = "open", + sort: Annotated[ + str | None, + Field( + description=( + "Sort order: 'oldest', 'newest', 'leastupdate', 'mostupdate', " + "'leastcomment', 'mostcomment', 'priority'. Omit for Gitea's default." + ), + ), + ] = None, + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, + limit: Annotated[int, Field(description="Items per page (max 50)")] = 30, +) -> list[dict[str, Any]]: + """List pull requests in a repository, optionally filtered by state. + + Returns Gitea PullRequest objects (not Issue-style — the PR endpoint + returns richer head/base/mergeable info than the issues endpoint does + even when issues are filtered to ``type=pulls``). + """ + params: dict[str, Any] = { + "state": state, + "page": page, + "limit": limit, + } + if sort is not None: + params["sort"] = sort + return await get_client().get_list( + f"/repos/{owner}/{repo}/pulls", + params=params, + ) + + +@mcp.tool(annotations=_READ_ONLY) +async def get_pull_request( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + pull_number: Annotated[int, Field(description="Pull request number (the #N in the URL)")], +) -> dict[str, Any]: + """Get a single pull request by number, including its discussion comments. + + Mirrors :func:`gitea_mcp.tools.issues.get_issue`'s shape: returns the full + Gitea PullRequest object with an additional ``comments_list`` field + containing the issue-style comment thread. Inline review comments (those + attached to specific diff lines) are NOT included — those live under + ``/pulls/{n}/reviews`` and are out of scope for this tool. + """ + client = get_client() + pr = await client.get_json(f"/repos/{owner}/{repo}/pulls/{pull_number}") + pr["comments_list"] = await client.get_list( + f"/repos/{owner}/{repo}/issues/{pull_number}/comments" + ) + return pr + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + openWorldHint=True, + ) +) +async def add_comment_on_pr( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + pull_number: Annotated[int, Field(description="Pull request number")], + body: Annotated[str, Field(description="Comment body (Markdown)")], +) -> dict[str, Any]: + """Add a comment to a pull request's discussion thread. + + Posts to ``/repos/{owner}/{repo}/issues/{pull_number}/comments`` — Gitea's + issue-comments endpoint also serves PR conversation comments (PRs and + issues share the number namespace and comment infrastructure). For inline + diff-line review comments, a separate ``/pulls/{n}/reviews``-based tool + would be needed; that's deliberately out of scope here. + """ + return await get_client().post_json( + f"/repos/{owner}/{repo}/issues/{pull_number}/comments", + json={"body": body}, + ) + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=True, + openWorldHint=True, + ) +) +async def merge_pr( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + pull_number: Annotated[int, Field(description="Pull request number")], + do: Annotated[ + str, + Field( + description=( + "Merge strategy: 'merge', 'rebase', 'rebase-merge', or 'squash'." + ) + ), + ] = "merge", + merge_title_field: Annotated[ + str | None, + Field( + description=( + "Optional merge commit title. Ignored by strategies that do not create " + "a merge commit." + ) + ), + ] = None, + merge_message_field: Annotated[ + str | None, + Field( + description=( + "Optional merge commit message body. Ignored by strategies that do not " + "create a merge commit." + ) + ), + ] = None, +) -> dict[str, Any]: + """Merge a pull request. + + Calls ``POST /repos/{owner}/{repo}/pulls/{pull_number}/merge`` using the + requested merge strategy and optional commit-message overrides. This is + tagged ``destructiveHint=True`` because it changes repository history and + closes the pull request. + """ + payload: dict[str, Any] = {"Do": do} + if merge_title_field is not None: + payload["MergeTitleField"] = merge_title_field + if merge_message_field is not None: + payload["MergeMessageField"] = merge_message_field + + return await get_client().post_json( + f"/repos/{owner}/{repo}/pulls/{pull_number}/merge", + json=payload, + ) diff --git a/package/gitea_mcp/tools/releases.py b/package/gitea_mcp/tools/releases.py new file mode 100644 index 0000000..66533bd --- /dev/null +++ b/package/gitea_mcp/tools/releases.py @@ -0,0 +1,97 @@ +"""MCP tools for Gitea releases.""" + +from __future__ import annotations + +from typing import Annotated, Any + +from mcp.types import ToolAnnotations +from pydantic import Field + +from gitea_mcp._app import get_client, mcp + + +@mcp.tool( + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True) +) +async def list_releases( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, + limit: Annotated[int, Field(description="Items per page (max 50)")] = 30, +) -> list[dict[str, Any]]: + """List releases for a repository. + + Returns the standard Gitea Release object array, including drafts and + pre-releases. Sort order is newest first. + """ + return await get_client().get_list( + f"/repos/{owner}/{repo}/releases", + params={"page": page, "limit": limit}, + ) + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + # Creates a release AND (per docstring warning) creates the underlying + # git tag if it doesn't exist. That tag-creation side effect is what + # makes this not idempotent (re-running with same tag_name errors). + destructiveHint=False, + idempotentHint=False, + openWorldHint=True, + ) +) +async def create_release( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + tag_name: Annotated[ + str, + Field( + description=( + "Tag this release is based on. If the tag does not already exist " + "in the repository, Gitea creates it at the time of release." + ), + ), + ], + name: Annotated[str, Field(description="Release title")], + body: Annotated[str, Field(description="Release notes in Markdown")] = "", + target_commitish: Annotated[ + str | None, + Field( + description=( + "Branch name or commit SHA the tag should point at. " + "Defaults to the repository's default branch. " + "Ignored if the tag already exists." + ), + ), + ] = None, + draft: Annotated[ + bool, + Field(description="Save as draft without publishing"), + ] = False, + prerelease: Annotated[ + bool, + Field(description="Mark as a pre-release"), + ] = False, +) -> dict[str, Any]: + """Create a new release in a repository. + + .. warning:: + + Side effects: if ``tag_name`` does not already exist in the repository, + Gitea creates the tag at the current ``target_commitish`` (or default + branch). Creating a draft does NOT skip tag creation — both drafts and + published releases will leave a tag in the repo. + """ + payload: dict[str, Any] = { + "tag_name": tag_name, + "name": name, + "body": body, + "draft": draft, + "prerelease": prerelease, + } + if target_commitish is not None: + payload["target_commitish"] = target_commitish + return await get_client().post_json( + f"/repos/{owner}/{repo}/releases", json=payload + ) diff --git a/package/gitea_mcp/tools/repos.py b/package/gitea_mcp/tools/repos.py new file mode 100644 index 0000000..17f5d0b --- /dev/null +++ b/package/gitea_mcp/tools/repos.py @@ -0,0 +1,90 @@ +"""MCP tools for Gitea repository metadata.""" + +from __future__ import annotations + +from typing import Annotated, Any + +from mcp.types import ToolAnnotations +from pydantic import Field + +from gitea_mcp._app import get_client, mcp +from gitea_mcp.client import GiteaAPIError + +_READ_ONLY = ToolAnnotations(readOnlyHint=True, openWorldHint=True) + + +@mcp.tool(annotations=_READ_ONLY) +async def list_repos( + owner: Annotated[ + str | None, + Field( + description=( + "Username or organization to list repos for. " + "Leave empty to list repositories accessible to the authenticated user." + ), + ), + ] = None, + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, + limit: Annotated[int, Field(description="Items per page (max 50)")] = 30, +) -> list[dict[str, Any]]: + """List repositories. + + Three modes: + + - ``owner`` empty: returns repositories accessible to the authenticated user + (``GET /user/repos``). + - ``owner`` is a user: returns that user's repositories + (``GET /users/{owner}/repos``). + - ``owner`` is an organization: returns that org's repositories + (``GET /orgs/{owner}/repos`` — automatically tried as a fallback when the + user endpoint 404s, so callers don't need to know which it is). + """ + client = get_client() + params: dict[str, Any] = {"page": page, "limit": limit} + + if not owner: + return await client.get_list("/user/repos", params=params) + + try: + return await client.get_list(f"/users/{owner}/repos", params=params) + except GiteaAPIError as e: + if e.status_code == 404: + # Owner is likely an organization — fall back transparently. + return await client.get_list(f"/orgs/{owner}/repos", params=params) + raise + + +@mcp.tool(annotations=_READ_ONLY) +async def list_labels( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, + limit: Annotated[int, Field(description="Items per page (max 50)")] = 30, +) -> list[dict[str, Any]]: + """List labels defined in a repository. + + Returns the standard Gitea Label object: ``{id, name, color, description, ...}``. + Use the ``id`` values when calling tools that take ``label_ids`` directly; + most tools accept label *names* and resolve to IDs internally. + """ + return await get_client().get_list( + f"/repos/{owner}/{repo}/labels", + params={"page": page, "limit": limit}, + ) + + +@mcp.tool(annotations=_READ_ONLY) +async def list_milestones( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + state: Annotated[ + str, Field(description="Filter by state: 'open', 'closed', or 'all'") + ] = "open", + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, + limit: Annotated[int, Field(description="Items per page (max 50)")] = 30, +) -> list[dict[str, Any]]: + """List milestones in a repository, optionally filtered by state.""" + return await get_client().get_list( + f"/repos/{owner}/{repo}/milestones", + params={"state": state, "page": page, "limit": limit}, + ) diff --git a/package/gitea_mcp/tools/server_info.py b/package/gitea_mcp/tools/server_info.py new file mode 100644 index 0000000..daf059c --- /dev/null +++ b/package/gitea_mcp/tools/server_info.py @@ -0,0 +1,80 @@ +"""MCP tools for introspecting the running gitea-mcp server. + +These are *runtime* introspection tools — callable from any MCP client (e.g. +Claude during a conversation) to ask the server about itself. Different from +the ``gitea-mcp doctor`` CLI command, which is a pre-launch check the human +runs in a shell before pointing an MCP client at the server. + +Why this exists: + +* **Multi-instance disambiguation.** A user may have several gitea-mcp + servers configured in their MCP client (e.g. personal Gitea + work Gitea). + Calling ``get_server_info`` lets Claude tell them apart by URL and user. +* **Version-aware debugging.** When a behavior diverges from the documented + contract, the LLM can ask the server its version directly rather than + guessing. +* **Self-awareness for upgrade prompts.** A future client can compare the + running version against the latest on PyPI to nudge users to upgrade. +""" + +from __future__ import annotations + +from typing import Any + +from mcp.types import ToolAnnotations + +from gitea_mcp import __version__ +from gitea_mcp._app import get_client, mcp +from gitea_mcp.client import GiteaAPIError + +_READ_ONLY = ToolAnnotations(readOnlyHint=True, openWorldHint=True) + + +@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False)) +async def get_server_version() -> dict[str, str]: + """Return the running gitea-mcp server's version. + + No network call — just reports the package version. ``openWorldHint`` is + ``False`` for this tool only, since it doesn't touch the Gitea instance + or any external system. + """ + return {"gitea_mcp_version": __version__} + + +@mcp.tool(annotations=_READ_ONLY) +async def get_server_info() -> dict[str, Any]: + """Return information about the running gitea-mcp server and its Gitea connection. + + Performs two Gitea API calls — ``GET /user`` (to identify the authenticated + user the PAT belongs to) and ``GET /version`` (to report the Gitea instance + version). If ``/version`` is unavailable (older Gitea, or the PAT lacks the + scope), ``gitea_version`` falls back to ``None`` rather than raising. + + Returns a dict with: + + * ``gitea_mcp_version`` — the gitea-mcp package version (e.g. ``"0.4.1"``). + * ``gitea_url`` — the Gitea base URL this server is configured against. + * ``gitea_user`` — the ``login`` of the authenticated Gitea user. + * ``gitea_version`` — the Gitea instance version string, or ``None`` if + unavailable. + """ + client = get_client() + user = await client.get_json("/user") + gitea_version: str | None = None + try: + version_payload = await client.get_json("/version") + raw_version = version_payload.get("version") + if isinstance(raw_version, str): + gitea_version = raw_version + except GiteaAPIError: + # /version isn't always available (older Gitea, or PAT lacks scope). + # Don't fail the whole introspection over it. + gitea_version = None + + login = user.get("login") + return { + "gitea_mcp_version": __version__, + "gitea_url": client.base_url, + "gitea_user": login if isinstance(login, str) else None, + "gitea_version": gitea_version, + } diff --git a/package/gitea_mcp/tools/wiki.py b/package/gitea_mcp/tools/wiki.py new file mode 100644 index 0000000..acdc132 --- /dev/null +++ b/package/gitea_mcp/tools/wiki.py @@ -0,0 +1,275 @@ +"""MCP tools for Gitea repository wikis.""" + +from __future__ import annotations + +import base64 +import binascii +from typing import Annotated, Any +from urllib.parse import quote + +from mcp.types import ToolAnnotations +from pydantic import Field + +from gitea_mcp._app import get_client, mcp +from gitea_mcp.client import GiteaAPIError + +_READ_ONLY = ToolAnnotations(readOnlyHint=True, openWorldHint=True) + + +def _wiki_page_path(owner: str, repo: str, page_name: str) -> str: + """Build a wiki-page API path without allowing page names to add segments.""" + encoded_page_name = quote(page_name, safe="") + return f"/repos/{owner}/{repo}/wiki/page/{encoded_page_name}" + + +async def _resolve_wiki_page_name(owner: str, repo: str, page_name: str) -> str: + """Resolve a display title to Gitea's canonical ``sub_url`` value.""" + resolved_name, _ = await _resolve_wiki_page(owner, repo, page_name) + return resolved_name + + +async def _resolve_wiki_page(owner: str, repo: str, page_name: str) -> tuple[str, str | None]: + """Resolve a page name and return its canonical name and display title.""" + client = get_client() + page_number = 1 + while True: + pages = await client.get_list( + f"/repos/{owner}/{repo}/wiki/pages", + params={"page": page_number, "limit": 50}, + ) + for page in pages: + if page.get("title") == page_name or page.get("sub_url") == page_name: + sub_url = page.get("sub_url") + title = page.get("title") + resolved_name = sub_url if isinstance(sub_url, str) and sub_url else page_name + resolved_title = title if isinstance(title, str) and title else None + return resolved_name, resolved_title + if len(pages) < 50: + return page_name, None + page_number += 1 + + +def _with_decoded_content(page: dict[str, Any]) -> dict[str, Any]: + """Add a best-effort UTF-8 ``text`` field to a Gitea WikiPage object.""" + text: str | None = None + content = page.get("content_base64") + if isinstance(content, str): + try: + text = base64.b64decode(content, validate=True).decode("utf-8") + except (binascii.Error, UnicodeDecodeError): + text = None + page["text"] = text + return page + + +@mcp.tool(annotations=_READ_ONLY) +async def list_wiki_pages( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, + limit: Annotated[int, Field(description="Items per page (max 50)")] = 30, +) -> list[dict[str, Any]]: + """List wiki pages in a repository. + + Returns Gitea ``WikiPageMetaData`` objects. Page content is intentionally + omitted by the list endpoint; use :func:`get_wiki_page` for Markdown. + """ + client = get_client() + try: + return await client.get_list( + f"/repos/{owner}/{repo}/wiki/pages", + params={"page": page, "limit": limit}, + ) + except GiteaAPIError as exc: + if exc.status_code != 404: + raise + repository = await client.get_json(f"/repos/{owner}/{repo}") + if repository.get("has_wiki") is True: + # Gitea returns 404 until the wiki git repository is initialized. + return [] + raise + + +@mcp.tool(annotations=_READ_ONLY) +async def get_wiki_page( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + page_name: Annotated[ + str, + Field(description="Wiki page name or title", min_length=1), + ], +) -> dict[str, Any]: + """Read one wiki page. + + Returns Gitea's complete ``WikiPage`` object plus ``text``, a best-effort + UTF-8 decoding of ``content_base64``. ``text`` is ``None`` if the content + is not valid Base64-encoded UTF-8. + """ + client = get_client() + try: + result = await client.get_json(_wiki_page_path(owner, repo, page_name)) + except GiteaAPIError as exc: + if exc.status_code != 404: + raise + resolved_name = await _resolve_wiki_page_name(owner, repo, page_name) + if resolved_name == page_name: + raise + result = await client.get_json(_wiki_page_path(owner, repo, resolved_name)) + return _with_decoded_content(result) + + +@mcp.tool(annotations=_READ_ONLY) +async def list_wiki_revisions( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + page_name: Annotated[ + str, + Field(description="Wiki page name or title", min_length=1), + ], + page: Annotated[int, Field(description="Page number (1-indexed)")] = 1, +) -> dict[str, Any]: + """List revision metadata for one wiki page. + + Returns Gitea's ``WikiCommitList`` object containing ``count`` and the + current page of ``commits``. Gitea 1.25 exposes a page parameter but no + configurable page-size parameter for this endpoint. + """ + encoded_page_name = quote(page_name, safe="") + return await get_client().get_json( + f"/repos/{owner}/{repo}/wiki/revisions/{encoded_page_name}", + params={"page": page}, + ) + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + idempotentHint=False, + openWorldHint=True, + ) +) +async def create_wiki_page( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + title: Annotated[str, Field(description="New wiki page title", min_length=1)], + content: Annotated[str, Field(description="Wiki page content as UTF-8 Markdown")], + message: Annotated[ + str, + Field(description="Optional commit message summarizing the change"), + ] = "", +) -> dict[str, Any]: + """Create a wiki page. + + The tool accepts plain UTF-8 Markdown and performs Gitea's required Base64 + encoding locally. Repeating the same request is not idempotent because the + page already exists after the first successful call. + """ + payload = { + "title": title, + "content_base64": base64.b64encode(content.encode("utf-8")).decode("ascii"), + "message": message, + } + result = await get_client().post_json( + f"/repos/{owner}/{repo}/wiki/new", + json=payload, + ) + return _with_decoded_content(result) + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=True, + idempotentHint=False, + openWorldHint=True, + ) +) +async def update_wiki_page( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + page_name: Annotated[ + str, + Field(description="Current wiki page name or title", min_length=1), + ], + content: Annotated[ + str, + Field(description="Complete replacement content as UTF-8 Markdown"), + ], + title: Annotated[ + str | None, + Field(description="Optional new page title; omit to keep the current title"), + ] = None, + message: Annotated[ + str, + Field(description="Optional commit message summarizing the change"), + ] = "", +) -> dict[str, Any]: + """Replace a wiki page's content and optionally rename it. + + ``content`` is deliberately required: callers must first read the page and + submit the complete desired Markdown, preventing an omitted field from + accidentally clearing content on Gitea versions with different defaults. + """ + payload: dict[str, Any] = { + "content_base64": base64.b64encode(content.encode("utf-8")).decode("ascii"), + "message": message, + } + client = get_client() + resolved_name = page_name + if title is None: + resolved_name, current_title = await _resolve_wiki_page(owner, repo, page_name) + # Gitea 1.25 renames a page to "unnamed" when PATCH omits title. + payload["title"] = current_title or page_name + else: + payload["title"] = title + try: + result = await client.patch_json( + _wiki_page_path(owner, repo, resolved_name), + json=payload, + ) + except GiteaAPIError as exc: + if exc.status_code != 404: + raise + resolved_name = await _resolve_wiki_page_name(owner, repo, page_name) + if resolved_name == page_name: + raise + result = await client.patch_json( + _wiki_page_path(owner, repo, resolved_name), + json=payload, + ) + return _with_decoded_content(result) + + +@mcp.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=True, + idempotentHint=True, + openWorldHint=True, + ) +) +async def delete_wiki_page( + owner: Annotated[str, Field(description="Repository owner")], + repo: Annotated[str, Field(description="Repository name")], + page_name: Annotated[ + str, + Field(description="Wiki page name or title to delete", min_length=1), + ], +) -> dict[str, Any]: + """Delete a wiki page. + + This is a destructive operation and Gitea returns no response body. The + returned object confirms which page name was sent after the API succeeds. + """ + client = get_client() + try: + await client.delete(_wiki_page_path(owner, repo, page_name)) + except GiteaAPIError as exc: + if exc.status_code != 404: + raise + resolved_name = await _resolve_wiki_page_name(owner, repo, page_name) + if resolved_name == page_name: + raise + await client.delete(_wiki_page_path(owner, repo, resolved_name)) + return {"deleted": True, "page_name": page_name} diff --git a/reference/gitea-mcp-hardened.ps1 b/reference/gitea-mcp-hardened.ps1 new file mode 100644 index 0000000..a9d50ba --- /dev/null +++ b/reference/gitea-mcp-hardened.ps1 @@ -0,0 +1,94 @@ +#!/usr/bin/env pwsh + +[CmdletBinding()] +param( + [string]$EnvFile = $( + if ($env:GITEA_ENV_FILE) { + $env:GITEA_ENV_FILE + } else { + Join-Path $HOME ".codex/gitea.env" + } + ), + [string]$Version = "0.5.1", + [switch]$CheckConfig, + [Parameter(ValueFromRemainingArguments = $true)] + [string[]]$ServerArgs +) + +$ErrorActionPreference = "Stop" +$utf8 = [System.Text.UTF8Encoding]::new($false) +[Console]::OutputEncoding = $utf8 +$OutputEncoding = $utf8 + +if (-not (Test-Path -LiteralPath $EnvFile -PathType Leaf)) { + throw "Gitea MCP 配置文件不存在：$EnvFile" +} + +$values = @{} +foreach ($rawLine in Get-Content -LiteralPath $EnvFile) { + $line = $rawLine.Trim() + if (-not $line -or $line.StartsWith("#")) { + continue + } + + $pair = $line -split "=", 2 + if ($pair.Count -ne 2) { + throw "Gitea MCP 配置行必须使用 KEY=VALUE 格式。" + } + + $values[$pair[0].Trim()] = $pair[1].Trim() +} + +foreach ($name in @("GITEA_URL", "GITEA_TOKEN")) { + if (-not $values.ContainsKey($name) -or [string]::IsNullOrWhiteSpace($values[$name])) { + throw "$name 未配置或为空。" + } +} + +try { + $giteaUri = [Uri]$values["GITEA_URL"] +} catch { + throw "GITEA_URL 不是有效 URL。" +} + +if ($giteaUri.Scheme -notin @("http", "https")) { + throw "GITEA_URL 只支持 http 或 https。" +} + +if ($giteaUri.AbsolutePath.Trim("/") -ne "") { + throw "GITEA_URL 必须填写实例根地址，不要包含 /api/v1；gitea-mcp 会自动追加 API 路径。" +} + +if ($giteaUri.Scheme -eq "http" -and $values["GITEA_ALLOW_INSECURE_HTTP"] -ne "1") { + throw "当前使用 HTTP。确认接受 Token 明文传输风险后，在私有配置中设置 GITEA_ALLOW_INSECURE_HTTP=1。" +} + +foreach ($entry in $values.GetEnumerator()) { + if ($entry.Key -like "GITEA_*") { + Set-Item -Path "Env:$($entry.Key)" -Value $entry.Value + } +} + +$noProxyEntries = @($env:NO_PROXY -split "," | ForEach-Object { $_.Trim() } | Where-Object { $_ }) +if ($noProxyEntries -notcontains $giteaUri.Host) { + $noProxyEntries += $giteaUri.Host +} +$env:NO_PROXY = $noProxyEntries -join "," +$env:no_proxy = $env:NO_PROXY + +if ($values["GITEA_DIRECT"] -eq "1") { + foreach ($proxyVariable in @("ALL_PROXY", "all_proxy", "HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy")) { + Remove-Item -Path "Env:$proxyVariable" -ErrorAction SilentlyContinue + } +} + +if ($CheckConfig) { + Write-Output "Gitea MCP 配置有效：URL=$($giteaUri.GetLeftPart([UriPartial]::Authority))，Token 已设置，版本=$Version。" + exit 0 +} + +$uvx = Get-Command uvx -ErrorAction Stop +$stderrLog = Join-Path ([IO.Path]::GetTempPath()) "gitea-mcp-$PID.stderr.log" +& $uvx.Source --from "gitea-mcp==$Version" gitea-mcp @ServerArgs 2>> $stderrLog +exit $LASTEXITCODE +