zhanggangyong/agent_management
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add CI/CD workflow capabilities to code_ai_agent

Browse Source 
Add Git, SSH, and K8s deploy tools enabling code_ai_agent to act as
an autonomous DevOps worker for the agent-manager project:
- tools/workspace.py: isolated /workspace/{task_id} per task
- tools/git_tools.py: clone, branch, status, write-file, commit-push
- tools/ssh_tools.py: SSH exec via paramiko (Azure VM support)
- tools/deploy_tools.py: K8s rollout restart/status via kubectl
- api_server.py: new routes /api/v1/git/*, /api/v1/ssh/*, /api/v1/deploy/k8s
- Dockerfile: add git, openssh-client, sshpass; pip install paramiko gitpython kubernetes
- requirements.txt: add paramiko, gitpython, kubernetes

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
elipitc
2026-03-27 19:44:08 +08:00
co-authored by Claude Sonnet 4.6
parent ee73763c89
commit 192e8a95bf
8 changed files with 602 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent_templates/agents/code_ai_agent/Dockerfile
+16 -8
View File
@@ -8,25 +8,33 @@ WORKDIR /app
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
# 安装系统依赖
# 安装系统依赖（含 git、ssh）
RUN apt-get update && apt-get install -y \
gcc \
git \
openssh-client \
sshpass \
curl \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
COPY agent_templates/agents/code_ai_agent/requirements.txt .
# 安装 Python 依赖
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir -r requirements.txt requests paramiko gitpython
# 复制应用代码
COPY . .
COPY agent_templates/agents/code_ai_agent/ /app/
COPY agent_templates/common/agent_callback_utils.py /app/common/
RUN touch /app/common/__init__.py
# 创建项目存储目录
RUN mkdir -p /tmp/projects
# 创建项目存储目录和工作空间
RUN mkdir -p /tmp/projects /workspace
# 安装curl用于健康检查
RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*
# 配置 git 全局设置
RUN git config --global user.email "code-ai-agent@taijiagnet.com" \
&& git config --global user.name "Code AI Agent" \
&& git config --global credential.helper store
# 暴露端口
EXPOSE 8000 8001
agent_templates/agents/code_ai_agent/requirements.txt
+3 -3
View File
@@ -2,9 +2,9 @@ pydantic-ai
httpx
mcp
fastmcp
fastapi
uvicorn[standard]
python-multipart
fastapi>=0.104.0
uvicorn[standard]>=0.24.0
python-multipart
paramiko
gitpython
kubernetes
agent_templates/agents/code_ai_agent/src/server/api_server.py
+237
View File
@@ -24,17 +24,36 @@ from .mcp_server import (
analyze_project,
create_code_file
)
# DevOps 工具
from .tools.workspace import create_workspace, delete_workspace, list_workspaces, workspace_exists
from .tools.git_tools import git_clone, git_branch, git_status, git_diff, git_commit_push, git_write_file
from .tools.ssh_tools import ssh_exec, ssh_test_connection
from .tools.deploy_tools import deploy_rollout_restart, deploy_rollout_status, get_deployment_status
try:
from common.agent_callback_utils import AgentCallbackHandler, CallbackContextManager
CALLBACK_ENABLED = True
except ImportError:
CALLBACK_ENABLED = False
AgentCallbackHandler = None
CallbackContextManager = None
# 配置
API_VERSION = "v1"
SERVER_NAME = "代码助手 Agent API"
POD_NAME = os.getenv("POD_NAME", "code-ai-agent")
USER_ID = os.getenv("USER_ID", "")
callback_handler: Optional[AgentCallbackHandler] = None
# 创建 FastAPI 应用
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
# 启动时初始化
global callback_handler
print(f"🚀 {SERVER_NAME} 启动中...")
if CALLBACK_ENABLED and AgentCallbackHandler:
callback_handler = AgentCallbackHandler(agent_name=POD_NAME, user_id=USER_ID)
yield
# 关闭时清理
print(f"🛑 {SERVER_NAME} 关闭中...")
@@ -407,6 +426,15 @@ async def handle_mcp_request(request_data: Dict[str, Any], session_id: Optional[
tool_name = params.get("name")
arguments = params.get("arguments", {})
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-mcp-{tool_name}-{request_id or uuid.uuid4().hex}"
) as ctx:
ctx.add_tool(tool_name)
result = await call_mcp_tool(tool_name, arguments, api_key=api_key)
else:
result = await call_mcp_tool(tool_name, arguments, api_key=api_key)
return {
@@ -518,6 +546,20 @@ async def api_generate_code(request: GenerateCodeRequest, api_key: str = Depends
根据自然语言需求生成高质量的代码
"""
try:
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-generate-{uuid.uuid4().hex}"
) as ctx:
ctx.add_tool("generate_code")
result = await call_mcp_tool('generate_code', {
'requirement': request.requirement,
'language': request.language,
'style': request.style,
'project_root': request.project_root
}, api_key=api_key)
else:
result = await call_mcp_tool('generate_code', {
'requirement': request.requirement,
'language': request.language,
@@ -544,6 +586,18 @@ async def api_refactor_code(request: RefactorCodeRequest, api_key: str = Depends
改进代码质量、性能和可维护性
"""
try:
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-refactor-{uuid.uuid4().hex}"
) as ctx:
ctx.add_tool("refactor_code")
result = await call_mcp_tool('refactor_code', {
'code_content': request.code_content,
'refactoring_goal': request.refactoring_goal
}, api_key=api_key)
else:
result = await call_mcp_tool('refactor_code', {
'code_content': request.code_content,
'refactoring_goal': request.refactoring_goal
@@ -568,6 +622,18 @@ async def api_review_code(request: ReviewCodeRequest, api_key: str = Depends(ver
发现潜在问题、bug 和改进建议
"""
try:
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-review-{uuid.uuid4().hex}"
) as ctx:
ctx.add_tool("review_code")
result = await call_mcp_tool('review_code', {
'code_content': request.code_content,
'file_path': request.file_path
}, api_key=api_key)
else:
result = await call_mcp_tool('review_code', {
'code_content': request.code_content,
'file_path': request.file_path
@@ -592,6 +658,19 @@ async def api_organize_code(request: OrganizeCodeRequest, api_key: str = Depends
智能分析代码并自动组织到合适的文件夹中
"""
try:
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-organize-{uuid.uuid4().hex}"
) as ctx:
ctx.add_tool("organize_code")
result = await call_mcp_tool('organize_code', {
'code_content': request.code_content,
'code_type': request.code_type,
'project_root': request.project_root
}, api_key=api_key)
else:
result = await call_mcp_tool('organize_code', {
'code_content': request.code_content,
'code_type': request.code_type,
@@ -617,6 +696,17 @@ async def api_classify_code(request: ClassifyCodeRequest, api_key: str = Depends
分析代码内容，确定其应该属于哪个类别/文件夹
"""
try:
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-classify-{uuid.uuid4().hex}"
) as ctx:
ctx.add_tool("classify_code")
result = await call_mcp_tool('classify_code', {
'code_content': request.code_content
}, api_key=api_key)
else:
result = await call_mcp_tool('classify_code', {
'code_content': request.code_content
}, api_key=api_key)
@@ -640,6 +730,18 @@ async def api_analyze_project(request: AnalyzeProjectRequest, api_key: str = Dep
分析项目结构，提供项目概览和改进建议
"""
try:
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-analyze-{uuid.uuid4().hex}"
) as ctx:
ctx.add_tool("analyze_project")
result = await call_mcp_tool('analyze_project', {
'project_root': request.project_root,
'max_depth': request.max_depth
}, api_key=api_key)
else:
result = await call_mcp_tool('analyze_project', {
'project_root': request.project_root,
'max_depth': request.max_depth
@@ -664,6 +766,17 @@ async def api_suggest_structure(request: SuggestStructureRequest, api_key: str =
根据项目描述，建议合理的文件夹结构
"""
try:
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-structure-{uuid.uuid4().hex}"
) as ctx:
ctx.add_tool("suggest_folder_structure")
result = await call_mcp_tool('suggest_folder_structure', {
'project_description': request.project_description
}, api_key=api_key)
else:
result = await call_mcp_tool('suggest_folder_structure', {
'project_description': request.project_description
}, api_key=api_key)
@@ -687,6 +800,20 @@ async def api_create_file(request: CreateFileRequest, api_key: str = Depends(ver
在指定文件夹中创建代码文件
"""
try:
if CALLBACK_ENABLED and callback_handler:
with CallbackContextManager(
handler=callback_handler,
user_id=USER_ID,
request_id=f"codeai-create-file-{uuid.uuid4().hex}"
) as ctx:
ctx.add_tool("create_code_file")
result = await call_mcp_tool('create_code_file', {
'code_content': request.code_content,
'folder_path': request.folder_path,
'file_name': request.file_name,
'project_root': request.project_root
}, api_key=api_key)
else:
result = await call_mcp_tool('create_code_file', {
'code_content': request.code_content,
'folder_path': request.folder_path,
@@ -705,6 +832,116 @@ async def api_create_file(request: CreateFileRequest, api_key: str = Depends(ver
)
# ==================== DevOps 请求模型 ====================
class GitCloneRequest(BaseModel):
repo_url: str = Field(default="http://gitee.ath.cx:3000/zhanggangyong/agent_management.git")
task_id: Optional[str] = None
branch: str = "master"
depth: int = 1
class GitBranchRequest(BaseModel):
task_id: str
branch_name: str
base_branch: str = "master"
class GitCommitPushRequest(BaseModel):
task_id: str
message: str
branch: str
token: Optional[str] = None
username: Optional[str] = None
class GitWriteFileRequest(BaseModel):
task_id: str
relative_path: str
content: str
class SshExecRequest(BaseModel):
command: str
host: Optional[str] = None
user: Optional[str] = None
password: Optional[str] = None
timeout: int = 120
class DeployRequest(BaseModel):
deployment: str = "agent-manager"
namespace: str = "agent-manager"
context: Optional[str] = None
wait: bool = True
# ==================== DevOps 路由 ====================
@app.post("/api/v1/git/clone")
async def api_git_clone(request: GitCloneRequest, api_key: str = Depends(verify_api_key)):
result = git_clone(request.repo_url, request.task_id or str(uuid.uuid4())[:8], request.branch, request.depth)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("stderr", "clone 失败"))
return APIResponse(success=True, data=result, message="Clone 成功")
@app.post("/api/v1/git/branch")
async def api_git_branch(request: GitBranchRequest, api_key: str = Depends(verify_api_key)):
result = git_branch(request.task_id, request.branch_name, request.base_branch)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("stderr", "创建分支失败"))
return APIResponse(success=True, data=result, message=f"分支 {request.branch_name} 创建成功")
@app.post("/api/v1/git/status")
async def api_git_status(request: dict, api_key: str = Depends(verify_api_key)):
task_id = request.get("task_id")
if not task_id:
raise HTTPException(status_code=400, detail="task_id 必填")
result = git_status(task_id)
return APIResponse(success=True, data=result, message="状态获取成功")
@app.post("/api/v1/git/write-file")
async def api_git_write_file(request: GitWriteFileRequest, api_key: str = Depends(verify_api_key)):
result = git_write_file(request.task_id, request.relative_path, request.content)
if not result.get("success"):
raise HTTPException(status_code=500, detail="写入文件失败")
return APIResponse(success=True, data=result, message="文件写入成功")
@app.post("/api/v1/git/commit-push")
async def api_git_commit_push(request: GitCommitPushRequest, api_key: str = Depends(verify_api_key)):
result = git_commit_push(request.task_id, request.message, request.branch,
token=request.token, username=request.username)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("stderr", "commit/push 失败"))
return APIResponse(success=True, data=result, message="Push 成功")
@app.post("/api/v1/ssh/exec")
async def api_ssh_exec(request: SshExecRequest, api_key: str = Depends(verify_api_key)):
result = ssh_exec(request.command, host=request.host, user=request.user,
password=request.password, timeout=request.timeout)
return APIResponse(success=result["success"], data=result,
message="命令执行成功" if result["success"] else "命令执行失败")
@app.post("/api/v1/ssh/test")
async def api_ssh_test(api_key: str = Depends(verify_api_key)):
result = ssh_test_connection()
return APIResponse(success=result["success"], data=result,
message="SSH 连接正常" if result["success"] else "SSH 连接失败")
@app.post("/api/v1/deploy/k8s")
async def api_deploy_k8s(request: DeployRequest, api_key: str = Depends(verify_api_key)):
result = deploy_rollout_restart(request.deployment, request.namespace, request.context)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("stderr", "部署触发失败"))
if request.wait:
status = deploy_rollout_status(request.deployment, request.namespace, request.context)
result["rollout_status"] = status
return APIResponse(success=True, data=result, message="部署成功")
@app.get("/api/v1/workspace")
async def api_list_workspaces(api_key: str = Depends(verify_api_key)):
return APIResponse(success=True, data={"workspaces": list_workspaces()}, message="获取成功")
@app.delete("/api/v1/workspace/{task_id}")
async def api_delete_workspace(task_id: str, api_key: str = Depends(verify_api_key)):
ok = delete_workspace(task_id)
return APIResponse(success=ok, data={"task_id": task_id}, message="工作空间已删除" if ok else "工作空间不存在")
if __name__ == '__main__':
import uvicorn
agent_templates/agents/code_ai_agent/src/server/tools/__init__.py
View File
agent_templates/agents/code_ai_agent/src/server/tools/deploy_tools.py
+67
View File
@@ -0,0 +1,67 @@
"""
部署工具 - 触发 K8s 滚动更新
"""
import os
import subprocess
import logging
from typing import Optional
logger = logging.getLogger(__name__)
K8S_CONTEXT = os.getenv("K8S_CONTEXT", "")
K8S_NAMESPACE = os.getenv("K8S_NAMESPACE", "agent-manager")
def _run_kubectl(cmd: list, timeout: int = 60) -> dict:
"""执行 kubectl 命令"""
try:
result = subprocess.run(
cmd, capture_output=True, text=True, timeout=timeout
)
return {
"returncode": result.returncode,
"stdout": result.stdout.strip(),
"stderr": result.stderr.strip(),
"success": result.returncode == 0,
}
except subprocess.TimeoutExpired:
return {"returncode": -1, "stdout": "", "stderr": "kubectl 命令超时", "success": False}
except Exception as e:
return {"returncode": -1, "stdout": "", "stderr": str(e), "success": False}
def deploy_rollout_restart(deployment: str, namespace: Optional[str] = None,
context: Optional[str] = None) -> dict:
"""触发 Deployment 滚动重启"""
ns = namespace or K8S_NAMESPACE
cmd = ["kubectl", "rollout", "restart", f"deployment/{deployment}", "-n", ns]
if context or K8S_CONTEXT:
cmd = ["kubectl", "--context", context or K8S_CONTEXT] + cmd[1:]
result = _run_kubectl(cmd)
if result["success"]:
logger.info(f"✅ 触发滚动重启: {deployment} in {ns}")
return result
def deploy_rollout_status(deployment: str, namespace: Optional[str] = None,
context: Optional[str] = None, timeout: int = 120) -> dict:
"""等待 Deployment 滚动更新完成"""
ns = namespace or K8S_NAMESPACE
cmd = ["kubectl", "rollout", "status", f"deployment/{deployment}",
"-n", ns, f"--timeout={timeout}s"]
if context or K8S_CONTEXT:
cmd = ["kubectl", "--context", context or K8S_CONTEXT] + cmd[1:]
result = _run_kubectl(cmd, timeout=timeout + 10)
if result["success"]:
logger.info(f"✅ 滚动更新完成: {deployment}")
return result
def get_deployment_status(deployment: str, namespace: Optional[str] = None,
context: Optional[str] = None) -> dict:
"""获取 Deployment 当前状态"""
ns = namespace or K8S_NAMESPACE
cmd = ["kubectl", "get", "deployment", deployment, "-n", ns, "-o", "json"]
if context or K8S_CONTEXT:
cmd = ["kubectl", "--context", context or K8S_CONTEXT] + cmd[1:]
return _run_kubectl(cmd)
agent_templates/agents/code_ai_agent/src/server/tools/git_tools.py
+135
View File
@@ -0,0 +1,135 @@
"""
Git 操作工具 - 支持 clone、branch、commit、push、status、diff
"""
import os
import subprocess
import logging
from typing import Optional
from .workspace import get_workspace_path, create_workspace, workspace_exists
logger = logging.getLogger(__name__)
GITEE_TOKEN = os.getenv("GITEE_TOKEN", "")
GITEE_USERNAME = os.getenv("GITEE_USERNAME", "")
def _run_git(cmd: list, cwd: str, timeout: int = 60) -> dict:
"""执行 git 命令，返回 stdout/stderr/returncode"""
try:
result = subprocess.run(
cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout
)
return {
"returncode": result.returncode,
"stdout": result.stdout.strip(),
"stderr": result.stderr.strip(),
"success": result.returncode == 0,
}
except subprocess.TimeoutExpired:
return {"returncode": -1, "stdout": "", "stderr": "命令超时", "success": False}
except Exception as e:
return {"returncode": -1, "stdout": "", "stderr": str(e), "success": False}
def _inject_credentials(repo_url: str) -> str:
"""将 token 注入到 git URL（不落盘）"""
if GITEE_TOKEN and "://" in repo_url:
proto, rest = repo_url.split("://", 1)
# 移除已有的凭证
if "@" in rest:
rest = rest.split("@", 1)[1]
return f"{proto}://{GITEE_USERNAME}:{GITEE_TOKEN}@{rest}"
return repo_url
def git_clone(repo_url: str, task_id: str, branch: str = "master", depth: int = 1) -> dict:
"""Clone 仓库到工作空间"""
if not workspace_exists(task_id):
create_workspace(task_id)
workspace = get_workspace_path(task_id)
repo_dir = os.path.join(workspace, "repo")
if os.path.exists(repo_dir):
return {"success": True, "message": "仓库已存在", "repo_dir": repo_dir, "task_id": task_id}
auth_url = _inject_credentials(repo_url)
cmd = ["git", "clone", "--branch", branch, "--depth", str(depth), auth_url, "repo"]
result = _run_git(cmd, cwd=workspace, timeout=120)
if result["success"]:
# 替换 remote URL 为无凭证版本（安全）
_run_git(["git", "remote", "set-url", "origin", repo_url], cwd=repo_dir)
# 配置 credential helper 使用 token
_run_git(["git", "config", "credential.helper", f"!echo password={GITEE_TOKEN}; echo username={GITEE_USERNAME}"], cwd=repo_dir)
result["repo_dir"] = repo_dir
result["task_id"] = task_id
logger.info(f"✅ Clone 成功: {repo_url} -> {repo_dir}")
else:
logger.error(f"❌ Clone 失败: {result['stderr']}")
return result
def git_branch(task_id: str, branch_name: str, base_branch: str = "master") -> dict:
"""创建并切换到新分支"""
repo_dir = os.path.join(get_workspace_path(task_id), "repo")
# 先确保在 base_branch
_run_git(["git", "checkout", base_branch], cwd=repo_dir)
result = _run_git(["git", "checkout", "-b", branch_name], cwd=repo_dir)
if result["success"]:
logger.info(f"✅ 创建分支: {branch_name}")
return result
def git_status(task_id: str) -> dict:
"""查看工作区状态"""
repo_dir = os.path.join(get_workspace_path(task_id), "repo")
result = _run_git(["git", "status", "--short"], cwd=repo_dir)
if result["success"]:
result["current_branch"] = _run_git(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=repo_dir)["stdout"]
return result
def git_diff(task_id: str) -> dict:
"""查看变更内容"""
repo_dir = os.path.join(get_workspace_path(task_id), "repo")
return _run_git(["git", "diff"], cwd=repo_dir)
def git_commit_push(task_id: str, message: str, branch: str,
token: Optional[str] = None, username: Optional[str] = None) -> dict:
"""Commit 并 Push 到远端"""
repo_dir = os.path.join(get_workspace_path(task_id), "repo")
# 暂存所有变更
add_result = _run_git(["git", "add", "-A"], cwd=repo_dir)
if not add_result["success"]:
return add_result
# Commit
commit_result = _run_git(["git", "commit", "-m", message], cwd=repo_dir)
if not commit_result["success"]:
return commit_result
# Push（使用 token 注入 URL）
origin_url = _run_git(["git", "remote", "get-url", "origin"], cwd=repo_dir)["stdout"]
# 优先使用请求中传入的 token
_token = token or GITEE_TOKEN
_username = username or GITEE_USERNAME
if _token and "://" in origin_url:
proto, rest = origin_url.split("://", 1)
if "@" in rest:
rest = rest.split("@", 1)[1]
auth_url = f"{proto}://{_username}:{_token}@{rest}"
else:
auth_url = _inject_credentials(origin_url)
push_result = _run_git(["git", "push", auth_url, branch], cwd=repo_dir, timeout=120)
if push_result["success"]:
logger.info(f"✅ Push 成功: {branch}")
else:
logger.error(f"❌ Push 失败: {push_result['stderr']}")
return push_result
def git_write_file(task_id: str, relative_path: str, content: str) -> dict:
"""在仓库内写入文件内容"""
repo_dir = os.path.join(get_workspace_path(task_id), "repo")
full_path = os.path.join(repo_dir, relative_path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, "w", encoding="utf-8") as f:
f.write(content)
return {"success": True, "path": full_path, "bytes": len(content.encode())}
agent_templates/agents/code_ai_agent/src/server/tools/ssh_tools.py
+58
View File
@@ -0,0 +1,58 @@
"""
SSH 工具 - 连接 Azure VM 执行命令
"""
import os
import logging
from typing import Optional
logger = logging.getLogger(__name__)
SSH_HOST = os.getenv("SSH_HOST", "")
SSH_USER = os.getenv("SSH_USER", "")
SSH_PASSWORD = os.getenv("SSH_PASSWORD", "")
SSH_PORT = int(os.getenv("SSH_PORT", "22"))
def ssh_exec(command: str, host: Optional[str] = None, user: Optional[str] = None,
password: Optional[str] = None, port: int = 22, timeout: int = 120) -> dict:
"""SSH 连接执行命令"""
try:
import paramiko
except ImportError:
return {"success": False, "stdout": "", "stderr": "paramiko 未安装", "returncode": -1}
_host = host or SSH_HOST
_user = user or SSH_USER
_password = password or SSH_PASSWORD
_port = port or SSH_PORT
if not _host or not _user:
return {"success": False, "stdout": "", "stderr": "SSH_HOST 或 SSH_USER 未配置", "returncode": -1}
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
client.connect(_host, port=_port, username=_user, password=_password, timeout=30)
stdin, stdout, stderr = client.exec_command(command, timeout=timeout)
stdout_str = stdout.read().decode("utf-8", errors="replace").strip()
stderr_str = stderr.read().decode("utf-8", errors="replace").strip()
returncode = stdout.channel.recv_exit_status()
logger.info(f"✅ SSH exec 完成 (rc={returncode}): {command[:80]}")
return {
"success": returncode == 0,
"stdout": stdout_str,
"stderr": stderr_str,
"returncode": returncode,
"host": _host,
}
except Exception as e:
logger.error(f"❌ SSH 连接失败: {e}")
return {"success": False, "stdout": "", "stderr": str(e), "returncode": -1}
finally:
client.close()
def ssh_test_connection(host: Optional[str] = None, user: Optional[str] = None,
password: Optional[str] = None) -> dict:
"""测试 SSH 连接是否正常"""
return ssh_exec("echo 'SSH_OK'", host=host, user=user, password=password, timeout=10)
agent_templates/agents/code_ai_agent/src/server/tools/workspace.py
+49
View File
@@ -0,0 +1,49 @@
"""
工作空间管理 - 为每个任务提供隔离的工作目录
"""
import os
import shutil
import uuid
from pathlib import Path
from typing import Optional
WORKSPACE_ROOT = os.getenv("WORKSPACE_ROOT", "/workspace")
def get_workspace_path(task_id: str) -> str:
"""获取任务工作空间路径"""
return os.path.join(WORKSPACE_ROOT, task_id)
def create_workspace(task_id: Optional[str] = None) -> str:
"""创建隔离工作空间，返回 task_id"""
if not task_id:
task_id = str(uuid.uuid4())[:8]
path = get_workspace_path(task_id)
os.makedirs(path, exist_ok=True)
return task_id
def delete_workspace(task_id: str) -> bool:
"""删除工作空间"""
path = get_workspace_path(task_id)
if os.path.exists(path):
shutil.rmtree(path)
return True
return False
def list_workspaces() -> list:
"""列出所有工作空间"""
root = Path(WORKSPACE_ROOT)
if not root.exists():
return []
return [
{"task_id": d.name, "path": str(d), "size_mb": round(sum(f.stat().st_size for f in d.rglob("*") if f.is_file()) / 1024 / 1024, 2)}
for d in root.iterdir() if d.is_dir()
]
def workspace_exists(task_id: str) -> bool:
"""检查工作空间是否存在"""
return os.path.exists(get_workspace_path(task_id))