103 lines
2.9 KiB
Python
103 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
发布Agent脚本
|
|
"""
|
|
import requests
|
|
import sys
|
|
|
|
BASE_URL = "http://localhost:8037"
|
|
|
|
def login(username="admin", password="123456"):
|
|
"""用户登录"""
|
|
login_data = {
|
|
"username": username,
|
|
"password": password
|
|
}
|
|
|
|
try:
|
|
response = requests.post(f"{BASE_URL}/api/v1/auth/login", data=login_data)
|
|
if response.status_code != 200:
|
|
print(f"❌ 登录失败: {response.status_code}")
|
|
return None, None
|
|
|
|
token = response.json().get("access_token")
|
|
if not token:
|
|
print("❌ 登录失败: 未获取到token")
|
|
return None, None
|
|
|
|
print(f"✅ 登录成功 (用户: {username})")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
return token, headers
|
|
except Exception as e:
|
|
print(f"❌ 登录异常: {str(e)}")
|
|
return None, None
|
|
|
|
def deploy_agent(agent_id, headers):
|
|
"""发布Agent"""
|
|
try:
|
|
response = requests.post(
|
|
f"{BASE_URL}/api/v1/agents/{agent_id}/deploy",
|
|
headers=headers
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
agent = response.json()
|
|
print(f"✅ Agent发布成功: {agent.get('name')} (状态: {agent.get('status')})")
|
|
return True
|
|
else:
|
|
print(f"❌ 发布失败: {response.status_code}")
|
|
print(f"响应: {response.text}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ 发布异常: {str(e)}")
|
|
return False
|
|
|
|
def find_agent_by_name(agent_name, headers):
|
|
"""通过名称查找Agent"""
|
|
try:
|
|
response = requests.get(
|
|
f"{BASE_URL}/api/v1/agents",
|
|
headers=headers,
|
|
params={"search": agent_name, "limit": 100}
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
print(f"❌ 获取Agent列表失败: {response.status_code}")
|
|
return None
|
|
|
|
agents = response.json()
|
|
for agent in agents:
|
|
if agent.get("name") == agent_name:
|
|
return agent
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"❌ 查找Agent异常: {str(e)}")
|
|
return None
|
|
|
|
if __name__ == "__main__":
|
|
agent_name = sys.argv[1] if len(sys.argv) > 1 else "知识库问答助手"
|
|
|
|
print(f"📝 发布Agent: {agent_name}\n")
|
|
|
|
# 登录
|
|
token, headers = login()
|
|
if not token:
|
|
sys.exit(1)
|
|
|
|
# 查找Agent
|
|
print(f"🔍 查找Agent: {agent_name}")
|
|
agent = find_agent_by_name(agent_name, headers)
|
|
if not agent:
|
|
print(f"❌ 未找到Agent: {agent_name}")
|
|
sys.exit(1)
|
|
|
|
print(f"✅ 找到Agent: {agent.get('name')} (ID: {agent.get('id')}, 状态: {agent.get('status')})")
|
|
|
|
# 发布Agent
|
|
print(f"\n🚀 发布Agent...")
|
|
if deploy_agent(agent.get('id'), headers):
|
|
print("\n✅ 完成！")
|
|
else:
|
|
sys.exit(1)
|