Initial commit
This commit is contained in:
QiuSW
@@ -0,0 +1,7 @@
|
||||
# Local credentials
|
||||
gitea.env
|
||||
|
||||
# Python runtime artifacts
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
|
||||
@@ -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 轮换后应更新独立的安全备份。
|
||||
@@ -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
|
||||
@@ -0,0 +1,3 @@
|
||||
GITEA_URL=https://gitea.example.com
|
||||
GITEA_TOKEN=your-personal-access-token
|
||||
|
||||
@@ -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 <samuel@waretech.services>
|
||||
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
|
||||
|
||||
<!-- mcp-name: io.github.werebear73/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).
|
||||
@@ -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,,
|
||||
@@ -0,0 +1,5 @@
|
||||
Wheel-Version: 1.0
|
||||
Generator: setuptools (83.0.0)
|
||||
Root-Is-Purelib: true
|
||||
Tag: py3-none-any
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
[console_scripts]
|
||||
gitea-mcp = gitea_mcp.server:main
|
||||
@@ -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.
|
||||
@@ -0,0 +1 @@
|
||||
gitea_mcp
|
||||
@@ -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__"]
|
||||
@@ -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
|
||||
@@ -0,0 +1 @@
|
||||
__version__ = '0.5.2.dev4'
|
||||
@@ -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 <PAT>`` 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
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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())
|
||||
@@ -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())
|
||||
@@ -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())
|
||||
@@ -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 <version>``.
|
||||
* ``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()
|
||||
@@ -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``.
|
||||
"""
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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},
|
||||
)
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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
|
||||
)
|
||||
@@ -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},
|
||||
)
|
||||
@@ -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,
|
||||
}
|
||||
@@ -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}
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user