Spaces:
Running
Running
Upload folder using huggingface_hub
Browse files- src/mcp-github/server.py +52 -1
src/mcp-github/server.py
CHANGED
|
@@ -23,7 +23,7 @@ except ImportError:
|
|
| 23 |
GithubException = Exception
|
| 24 |
|
| 25 |
# Initialize FastMCP Server
|
| 26 |
-
mcp = FastMCP("GitHub Operations")
|
| 27 |
|
| 28 |
def get_client():
|
| 29 |
token = os.environ.get("GITHUB_TOKEN")
|
|
@@ -173,6 +173,57 @@ def get_pull_request(owner: str, repo_name: str, pr_number: int) -> Dict[str, An
|
|
| 173 |
except Exception as e:
|
| 174 |
return {"error": str(e)}
|
| 175 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 176 |
if __name__ == "__main__":
|
| 177 |
import os
|
| 178 |
if os.environ.get("MCP_TRANSPORT") == "sse":
|
|
|
|
| 23 |
GithubException = Exception
|
| 24 |
|
| 25 |
# Initialize FastMCP Server
|
| 26 |
+
mcp = FastMCP("GitHub Operations", host="0.0.0.0")
|
| 27 |
|
| 28 |
def get_client():
|
| 29 |
token = os.environ.get("GITHUB_TOKEN")
|
|
|
|
| 173 |
except Exception as e:
|
| 174 |
return {"error": str(e)}
|
| 175 |
|
| 176 |
+
@mcp.tool()
|
| 177 |
+
def list_workflow_runs(owner: str, repo_name: str) -> List[Dict[str, Any]]:
|
| 178 |
+
"""
|
| 179 |
+
List recent workflow runs for a repository.
|
| 180 |
+
"""
|
| 181 |
+
log_usage("mcp-github", "list_workflow_runs")
|
| 182 |
+
try:
|
| 183 |
+
g = get_client()
|
| 184 |
+
repo = g.get_repo(f"{owner}/{repo_name}")
|
| 185 |
+
runs = repo.get_workflow_runs()
|
| 186 |
+
|
| 187 |
+
results = []
|
| 188 |
+
for run in runs[:10]:
|
| 189 |
+
results.append({
|
| 190 |
+
"id": run.id,
|
| 191 |
+
"name": run.name,
|
| 192 |
+
"status": run.status,
|
| 193 |
+
"conclusion": run.conclusion,
|
| 194 |
+
"event": run.event,
|
| 195 |
+
"created_at": str(run.created_at),
|
| 196 |
+
"url": run.html_url
|
| 197 |
+
})
|
| 198 |
+
return results
|
| 199 |
+
except Exception as e:
|
| 200 |
+
return [{"error": str(e)}]
|
| 201 |
+
|
| 202 |
+
@mcp.tool()
|
| 203 |
+
def get_workflow_run_details(owner: str, repo_name: str, run_id: int) -> Dict[str, Any]:
|
| 204 |
+
"""
|
| 205 |
+
Get details of a specific workflow run.
|
| 206 |
+
"""
|
| 207 |
+
log_usage("mcp-github", "get_workflow_run_details")
|
| 208 |
+
try:
|
| 209 |
+
g = get_client()
|
| 210 |
+
repo = g.get_repo(f"{owner}/{repo_name}")
|
| 211 |
+
run = repo.get_workflow_run(run_id)
|
| 212 |
+
|
| 213 |
+
return {
|
| 214 |
+
"id": run.id,
|
| 215 |
+
"name": run.name,
|
| 216 |
+
"status": run.status,
|
| 217 |
+
"conclusion": run.conclusion,
|
| 218 |
+
"event": run.event,
|
| 219 |
+
"created_at": str(run.created_at),
|
| 220 |
+
"updated_at": str(run.updated_at),
|
| 221 |
+
"url": run.html_url,
|
| 222 |
+
"run_number": run.run_number
|
| 223 |
+
}
|
| 224 |
+
except Exception as e:
|
| 225 |
+
return {"error": str(e)}
|
| 226 |
+
|
| 227 |
if __name__ == "__main__":
|
| 228 |
import os
|
| 229 |
if os.environ.get("MCP_TRANSPORT") == "sse":
|