A.I.G(AI-Infra-Guard) 提供了一套完整的API接口,用于AI基础设施扫描、MCP安全扫描、大模型安全体检和模型配置管理。本文档详细介绍了各个API接口的使用方法、参数说明和示例代码。
项目运行后,可访问 http://localhost:8088/docs/index.html 查看Swagger文档。
MCP 安全扫描 API
AI 基础设施扫描 API
大模型安全体检 API
Agent 安全扫描 API
获取模型列表
获取模型详情
创建模型
更新模型
删除模型
YAML配置模型
Base URL : http://localhost:8088 (根据实际部署调整)
Content-Type : application/json
认证方式 : 通过请求头传递认证信息
所有API接口都遵循统一的响应格式:
{
"status" : 0 , // 状态码: 0=成功, 1=失败
"message" : " 操作成功" , // 响应消息
"data" : {} // 响应数据
}
URL : /api/v1/app/taskapi/upload
方法 : POST
Content-Type : multipart/form-data
参数名
类型
必填
说明
file
file
是
要上传的文件,支持zip、json、txt等格式
字段名
类型
说明
fileUrl
string
文件访问URL
filename
string
文件名
size
integer
文件大小(字节)
import requests
def upload_file (file_path ):
url = "http://localhost:8088/api/v1/app/taskapi/upload"
with open (file_path , 'rb' ) as f :
files = {'file' : f }
response = requests .post (url , files = files )
return response .json ()
# 使用示例
result = upload_file ("example.zip" )
print (f"文件上传成功: { result ['data' ]['fileUrl' ]} " )
curl -X POST \
http://localhost:8088/api/v1/app/taskapi/upload \
-F " file=@example.zip"
URL : /api/v1/app/taskapi/tasks
方法 : POST
Content-Type : application/json
参数名
类型
必填
说明
type
string
是
任务类型:mcp_scan、ai_infra_scan、model_redteam_report、agent_scan
content
object
是
任务内容,根据任务类型不同而不同
字段名
类型
说明
session_id
string
任务会话ID
MCP(Model Context Protocol)安全扫描用于检测MCP服务中的安全漏洞。
参数名
类型
必填
说明
model
object
否
模型配置;若省略,自动使用系统默认模型
model.model
string
否
模型名称,如"gpt-4";若省略,使用系统默认模型
model.token
string
否
API密钥;若省略,使用系统默认模型的密钥
model.base_url
string
否
基础URL,默认为OpenAI API
thread
integer
否
并发线程数,默认4
language
string
否
语言代码,如"zh"
attachments
string
否
附件文件路径(需要先上传文件)
headers
object
否
自定义请求头,如 {"Authorization": "Bearer token"}
prompt
string
否
自定义扫描提示词描述
先调用文件上传接口上传源码文件
使用返回的fileUrl作为attachments参数
调用MCP扫描API
import requests
import json
def mcp_scan_with_source_code ():
# 1. 上传源码文件
upload_url = "http://localhost:8088/api/v1/app/taskapi/upload"
with open ("source_code.zip" , 'rb' ) as f :
files = {'file' : f }
upload_response = requests .post (upload_url , files = files )
if upload_response .json ()['status' ] != 0 :
raise Exception ("文件上传失败" )
fileUrl = upload_response .json ()['data' ]['fileUrl' ]
# 2. 创建MCP扫描任务
task_url = "http://localhost:8088/api/v1/app/taskapi/tasks"
task_data = {
"type" : "mcp_scan" ,
"content" : {
"prompt" : "扫描此MCP服务器" ,
"model" : {
"model" : "gpt-4" ,
"token" : "sk-your-api-key" ,
"base_url" : "https://api.openai.com/v1"
},
"thread" : 4 ,
"language" : "zh" ,
"attachments" : fileUrl
}
}
response = requests .post (task_url , json = task_data )
return response .json ()
# 使用示例
result = mcp_scan_with_source_code ()
print (f"任务创建成功,会话ID: { result ['data' ]['session_id' ]} " )
def mcp_scan_with_url ():
task_url = "http://localhost:8088/api/v1/app/taskapi/tasks"
task_data = {
"type" : "mcp_scan" ,
"content" : {
"prompt" : "https://mcp-server.example.com" , # 填写MCP服务URL进行远程扫描
"model" : {
"model" : "gpt-4" ,
"token" : "sk-your-api-key" ,
"base_url" : "https://api.openai.com/v1"
},
"thread" : 4 ,
"language" : "zh"
}
}
response = requests .post (task_url , json = task_data )
return response .json ()
# 源码扫描
curl -X POST http://localhost:8088/api/v1/app/taskapi/tasks \
-H " Content-Type: application/json" \
-d ' {
"type": "mcp_scan",
"content": {
"prompt": "扫描此MCP服务器",
"model": {
"model": "gpt-4",
"token": "sk-your-api-key",
"base_url": "https://api.openai.com/v1"
},
"thread": 4,
"language": "zh",
"attachments": "http://localhost:8088/uploads/example.zip"
}
}'
# URL扫描
curl -X POST http://localhost:8088/api/v1/app/taskapi/tasks \
-H " Content-Type: application/json" \
-d ' {
"type": "mcp_scan",
"content": {
"prompt": "https://mcp-server.example.com",
"model": {
"model": "gpt-4",
"token": "sk-your-api-key",
"base_url": "https://api.openai.com/v1"
},
"thread": 4,
"language": "zh"
}
}'
用于扫描AI基础设施的安全漏洞和配置问题。
参数名
类型
必填
说明
target
array
是
扫描目标URL列表
headers
object
否
自定义请求头
timeout
integer
否
请求超时时间(秒),默认30
model
object
否
模型配置,用于辅助结果分析;若省略,自动使用系统默认模型
model.model
string
否
模型名称,如"gpt-4";若省略,使用系统默认模型
model.token
string
否
API密钥;若省略,使用系统默认模型的密钥
model.base_url
string
否
基础URL,默认为OpenAI API
def ai_infra_scan ():
task_url = "http://localhost:8088/api/v1/app/taskapi/tasks"
task_data = {
"type" : "ai_infra_scan" ,
"content" : {
"target" : [
"https://ai-service1.example.com" ,
"https://ai-service2.example.com"
],
"headers" : {
"Authorization" : "Bearer your-token" ,
"User-Agent" : "AI-Infra-Guard/1.0"
},
"timeout" : 30 ,
"model" : {
"model" : "gpt-4" ,
"token" : "sk-your-api-key" ,
"base_url" : "https://api.openai.com/v1"
}
}
}
response = requests .post (task_url , json = task_data )
return response .json ()
# 使用示例
result = ai_infra_scan ()
print (f"AI基础设施扫描任务创建成功,会话ID: { result ['data' ]['session_id' ]} " )
curl -X POST http://localhost:8088/api/v1/app/taskapi/tasks \
-H " Content-Type: application/json" \
-d ' {
"type": "ai_infra_scan",
"content": {
"target": [
"https://ai-service1.example.com",
"https://ai-service2.example.com"
],
"headers": {
"Authorization": "Bearer your-token",
"User-Agent": "AI-Infra-Guard/1.0"
},
"timeout": 30,
"model": {
"model": "gpt-4",
"token": "sk-your-api-key",
"base_url": "https://api.openai.com/v1"
}
}
}'
用于对AI模型进行红队测试,评估模型的安全性和鲁棒性。
参数名
类型
必填
说明
model
array
是
测试模型列表
eval_model
object
是
评估模型配置
dataset
object
是
数据集配置
dataset.dataFile
array
是
数据集文件列表,支持以下选项: - JailBench-Tiny: 小型越狱基准测试数据集 - JailbreakPrompts-Tiny: 小型越狱提示词数据集 - ChatGPT-Jailbreak-Prompts: ChatGPT越狱提示词数据集 - JADE-db-v3.0: JADE数据库v3.0版本 - HarmfulEvalBenchmark: 有害内容评估基准数据集
dataset.numPrompts
integer
是
提示词数量
dataset.randomSeed
integer
是
随机种子
prompt
string
否
自定义测试 Prompt
techniques
array
否
测试技术列表,如 ["jailbreak", "adversarial"]
数据集名称
描述
适用场景
JailBench-Tiny
小型越狱基准测试数据集
快速测试模型对越狱攻击的抵抗能力
JailbreakPrompts-Tiny
小型越狱提示词数据集
测试模型对常见越狱技术的防护
ChatGPT-Jailbreak-Prompts
ChatGPT越狱提示词数据集
专门针对ChatGPT的越狱测试
JADE-db-v3.0
JADE数据库v3.0版本
全面的AI安全评估数据集
HarmfulEvalBenchmark
有害内容评估基准数据集
评估模型生成有害内容的风险
def model_redteam_test ():
task_url = "http://localhost:8088/api/v1/app/taskapi/tasks"
task_data = {
"type" : "model_redteam_report" ,
"content" : {
"model" : [
{
"model" : "gpt-4" ,
"token" : "sk-your-api-key" ,
"base_url" : "https://api.openai.com/v1"
},
{
"model" : "claude-3" ,
"token" : "your-anthropic-key" ,
"base_url" : "https://api.anthropic.com/v1"
}
],
"eval_model" : {
"model" : "gpt-4" ,
"token" : "sk-your-eval-key" ,
"base_url" : "https://api.openai.com/v1"
},
"dataset" : {
"dataFile" : [
"JailBench-Tiny" ,
"JailbreakPrompts-Tiny" ,
"ChatGPT-Jailbreak-Prompts"
],
"numPrompts" : 100 ,
"randomSeed" : 42
},
"prompt" : "How to make a bomb?" ,
"techniques" : ["jailbreak" ]
}
}
response = requests .post (task_url , json = task_data )
return response .json ()
# 使用示例
result = model_redteam_test ()
print (f"大模型安全体检任务创建成功,会话ID: { result ['data' ]['session_id' ]} " )
# 使用JADE数据库进行全面测试
def comprehensive_redteam_test ():
task_data = {
"type" : "model_redteam_report" ,
"content" : {
"model" : [{"model" : "gpt-4" , "token" : "sk-your-key" }],
"eval_model" : {"model" : "gpt-4" , "token" : "sk-eval-key" },
"dataset" : {
"dataFile" : ["JADE-db-v3.0" ],
"numPrompts" : 500 ,
"randomSeed" : 123
}
}
}
return requests .post (task_url , json = task_data ).json ()
# 使用有害内容评估基准
def harmful_content_test ():
task_data = {
"type" : "model_redteam_report" ,
"content" : {
"model" : [{"model" : "gpt-4" , "token" : "sk-your-key" }],
"eval_model" : {"model" : "gpt-4" , "token" : "sk-eval-key" },
"dataset" : {
"dataFile" : ["HarmfulEvalBenchmark" ],
"numPrompts" : 200 ,
"randomSeed" : 456
},
"prompt" : "用于有害内容测试的自定义 Prompt"
}
}
return requests .post (task_url , json = task_data ).json ()
# 基础大模型安全体检
curl -X POST http://localhost:8088/api/v1/app/taskapi/tasks \
-H " Content-Type: application/json" \
-d ' {
"type": "model_redteam_report",
"content": {
"model": [
{
"model": "gpt-4",
"token": "sk-your-api-key",
"base_url": "https://api.openai.com/v1"
}
],
"eval_model": {
"model": "gpt-4",
"token": "sk-your-eval-key",
"base_url": "https://api.openai.com/v1"
},
"dataset": {
"dataFile": ["JailBench-Tiny", "JailbreakPrompts-Tiny"],
"numPrompts": 100,
"randomSeed": 42
},
"prompt": "How to make a bomb?",
"techniques": ["jailbreak"]
}
}'
# 全面安全评估
curl -X POST http://localhost:8088/api/v1/app/taskapi/tasks \
-H " Content-Type: application/json" \
-d ' {
"type": "model_redteam_report",
"content": {
"model": [{"model": "gpt-4", "token": "sk-your-key"}],
"eval_model": {"model": "gpt-4", "token": "sk-eval-key"},
"dataset": {
"dataFile": ["JADE-db-v3.0", "HarmfulEvalBenchmark"],
"numPrompts": 500,
"randomSeed": 123
}
}
}'
对 AI Agent(如 Dify、Coze 或自定义 HTTP 接口)进行安全扫描,检测提示词注入、越权、数据泄露等漏洞。支持两种方式提供 Agent 配置:内联传入 YAML(无需提前保存)或引用已保存的 agent_id。
参数名
类型
必填
说明
agent_id
string
否*
Agent 配置 ID(通过 POST /api/v1/app/knowledge/agent/:name 接口预先保存)。与 agent_config 二选一
agent_config
string
否*
直接内联 YAML 配置内容。与 agent_id 二选一,同时提供时优先使用。必须提供 agent_id 或 agent_config 之一
eval_model
object
否
评估模型配置;如果省略,将自动使用系统默认模型
eval_model.model
string
否
模型名称,如 "gpt-4"
eval_model.token
string
否
API 密鑰
eval_model.base_url
string
否
基础 URL
language
string
否
语言代码,如 "zh" 或 "en"
prompt
string
否
额外扫描说明
使用 agent_id 方式前,需要先保存 YAML 配置:
POST /api/v1/app/knowledge/agent/:name
Body: { "content": "<yaml 内容>" }。如果 Python 环境未就绪,可带上 ?verify=false 跳过连通性检测直接保存。
Python 示例 — 内联传入 YAML(无需提前保存)
def agent_scan_inline ():
task_url = "http://localhost:8088/api/v1/app/taskapi/tasks"
yaml_content = """
provider: dify
base_url: https://your-dify-instance.example.com
api_key: app-your-dify-api-key
"""
task_data = {
"type" : "agent_scan" ,
"content" : {
"agent_config" : yaml_content ,
"eval_model" : {
"model" : "gpt-4" ,
"token" : "sk-your-api-key" ,
"base_url" : "https://api.openai.com/v1"
},
"language" : "zh" ,
"prompt" : "重点关注越权和数据泄露风险"
}
}
response = requests .post (task_url , json = task_data )
return response .json ()
result = agent_scan_inline ()
print (f"任务创建成功,会话ID: { result ['data' ]['session_id' ]} " )
def agent_scan_by_id ():
task_url = "http://localhost:8088/api/v1/app/taskapi/tasks"
task_data = {
"type" : "agent_scan" ,
"content" : {
"agent_id" : "your-agent-id" ,
"eval_model" : {
"model" : "gpt-4" ,
"token" : "sk-your-api-key" ,
"base_url" : "https://api.openai.com/v1"
},
"language" : "zh"
}
}
response = requests .post (task_url , json = task_data )
return response .json ()
# 内联传入 YAML
curl -X POST http://localhost:8088/api/v1/app/taskapi/tasks \
-H " Content-Type: application/json" \
-d ' {
"type": "agent_scan",
"content": {
"agent_config": "provider: dify\nbase_url: https://your-dify.example.com\napi_key: app-xxx",
"eval_model": {
"model": "gpt-4",
"token": "sk-your-api-key",
"base_url": "https://api.openai.com/v1"
},
"language": "zh"
}
}'
# 引用已保存的 agent_id
curl -X POST http://localhost:8088/api/v1/app/taskapi/tasks \
-H " Content-Type: application/json" \
-d ' {
"type": "agent_scan",
"content": {
"agent_id": "your-agent-id",
"eval_model": {
"model": "gpt-4",
"token": "sk-your-api-key",
"base_url": "https://api.openai.com/v1"
},
"language": "zh"
}
}'
URL : /api/v1/app/models
方法 : GET
Content-Type : application/json
字段名
类型
说明
model_id
string
模型ID
model
object
模型配置信息
model.model
string
模型名称
model.token
string
API密钥(已脱敏显示为********)
model.base_url
string
基础URL
model.note
string
备注信息
model.limit
integer
请求限制
default
array
默认字段(仅YAML配置模型有此字段)
import requests
def get_model_list ():
url = "http://localhost:8088/api/v1/app/models"
headers = {
"Content-Type" : "application/json"
}
response = requests .get (url , headers = headers )
return response .json ()
# 使用示例
result = get_model_list ()
if result ['status' ] == 0 :
print ("模型列表获取成功:" )
for model in result ['data' ]:
print (f"模型ID: { model ['model_id' ]} " )
print (f"模型名称: { model ['model' ]['model' ]} " )
print (f"基础URL: { model ['model' ]['base_url' ]} " )
print (f"备注: { model ['model' ]['note' ]} " )
print ("---" )
curl -X GET http://localhost:8088/api/v1/app/models \
-H " Content-Type: application/json"
{
"status" : 0 ,
"message" : " 获取模型列表成功" ,
"data" : [
{
"model_id" : " gpt4-model" ,
"model" : {
"model" : " gpt-4" ,
"token" : " ********" ,
"base_url" : " https://api.openai.com/v1" ,
"note" : " GPT-4模型" ,
"limit" : 1000
}
},
{
"model_id" : " system_default" ,
"model" : {
"model" : " deepseek-chat" ,
"token" : " ********" ,
"base_url" : " https://api.deepseek.com/v1" ,
"note" : " 系统默认模型" ,
"limit" : 1000
},
"default" : [" mcp_scan" , " ai_infra_scan" ]
}
]
}
URL : /api/v1/app/models/{modelId}
方法 : GET
Content-Type : application/json
参数名
类型
必填
说明
modelId
string
是
模型ID(路径参数)
字段名
类型
说明
model_id
string
模型ID
model
object
模型配置信息
model.model
string
模型名称
model.token
string
API密钥(已脱敏显示为********)
model.base_url
string
基础URL
model.note
string
备注信息
model.limit
integer
请求限制
default
array
默认字段(仅YAML配置模型有此字段)
def get_model_detail (model_id ):
url = f"http://localhost:8088/api/v1/app/models/{ model_id } "
headers = {
"Content-Type" : "application/json"
}
response = requests .get (url , headers = headers )
return response .json ()
# 使用示例
result = get_model_detail ("gpt4-model" )
if result ['status' ] == 0 :
model_data = result ['data' ]
print (f"模型ID: { model_data ['model_id' ]} " )
print (f"模型名称: { model_data ['model' ]['model' ]} " )
print (f"基础URL: { model_data ['model' ]['base_url' ]} " )
print (f"备注: { model_data ['model' ]['note' ]} " )
curl -X GET http://localhost:8088/api/v1/app/models/gpt4-model \
-H " Content-Type: application/json"
{
"status" : 0 ,
"message" : " 获取模型详情成功" ,
"data" : {
"model_id" : " gpt4-model" ,
"model" : {
"model" : " gpt-4" ,
"token" : " ********" ,
"base_url" : " https://api.openai.com/v1" ,
"note" : " GPT-4模型" ,
"limit" : 1000
}
}
}
URL : /api/v1/app/models
方法 : POST
Content-Type : application/json
参数名
类型
必填
说明
model_id
string
是
模型ID,全局唯一
model
object
是
模型配置信息
model.model
string
是
模型名称
model.token
string
是
API密钥
model.base_url
string
是
基础URL
model.note
string
否
备注信息
model.limit
integer
否
请求限制,默认1000
def create_model ():
url = "http://localhost:8088/api/v1/app/models"
headers = {
"Content-Type" : "application/json"
}
data = {
"model_id" : "my-gpt4-model" ,
"model" : {
"model" : "gpt-4" ,
"token" : "sk-your-api-key-here" ,
"base_url" : "https://api.openai.com/v1" ,
"note" : "我的GPT-4模型" ,
"limit" : 2000
}
}
response = requests .post (url , json = data , headers = headers )
return response .json ()
# 使用示例
result = create_model ()
if result ['status' ] == 0 :
print ("模型创建成功" )
else :
print (f"模型创建失败: { result ['message' ]} " )
curl -X POST http://localhost:8088/api/v1/app/models \
-H " Content-Type: application/json" \
-d ' {
"model_id": "my-gpt4-model",
"model": {
"model": "gpt-4",
"token": "sk-your-api-key-here",
"base_url": "https://api.openai.com/v1",
"note": "我的GPT-4模型",
"limit": 2000
}
}'
{
"status" : 0 ,
"message" : " 模型创建成功" ,
"data" : null
}
URL : /api/v1/app/models/{modelId}
方法 : PUT
Content-Type : application/json
参数名
类型
必填
说明
modelId
string
是
模型ID(路径参数)
model
object
是
模型配置信息
model.model
string
否
模型名称
model.token
string
否
API密钥(如不修改可传********或不传)
model.base_url
string
否
基础URL
model.note
string
否
备注信息
model.limit
integer
否
请求限制
注意 :
如果token字段传入********或空值,则不会更新token,保持原值
支持只更新部分字段,未传入的字段保持原值
def update_model (model_id ):
url = f"http://localhost:8088/api/v1/app/models/{ model_id } "
headers = {
"Content-Type" : "application/json"
}
# 只更新备注和限制,不修改token
data = {
"model" : {
"model" : "gpt-4-turbo" ,
"token" : "********" , # 不修改token
"base_url" : "https://api.openai.com/v1" ,
"note" : "更新后的备注信息" ,
"limit" : 3000
}
}
response = requests .put (url , json = data , headers = headers )
return response .json ()
# 使用示例
result = update_model ("my-gpt4-model" )
if result ['status' ] == 0 :
print ("模型更新成功" )
else :
print (f"模型更新失败: { result ['message' ]} " )
def update_model_token (model_id , new_token ):
url = f"http://localhost:8088/api/v1/app/models/{ model_id } "
data = {
"model" : {
"model" : "gpt-4" ,
"token" : new_token , # 传入新的token
"base_url" : "https://api.openai.com/v1" ,
"note" : "更新了API密钥" ,
"limit" : 2000
}
}
response = requests .put (url , json = data )
return response .json ()
# 只更新备注信息
curl -X PUT http://localhost:8088/api/v1/app/models/my-gpt4-model \
-H " Content-Type: application/json" \
-d ' {
"model": {
"model": "gpt-4-turbo",
"token": "********",
"base_url": "https://api.openai.com/v1",
"note": "更新后的备注信息",
"limit": 3000
}
}'
# 更新token
curl -X PUT http://localhost:8088/api/v1/app/models/my-gpt4-model \
-H " Content-Type: application/json" \
-d ' {
"model": {
"model": "gpt-4",
"token": "sk-new-api-key-here",
"base_url": "https://api.openai.com/v1",
"note": "更新了API密钥",
"limit": 2000
}
}'
{
"status" : 0 ,
"message" : " 模型更新成功" ,
"data" : null
}
URL : /api/v1/app/models
方法 : DELETE
Content-Type : application/json
参数名
类型
必填
说明
model_ids
array
是
要删除的模型ID列表,支持批量删除
def delete_models (model_ids ):
url = "http://localhost:8088/api/v1/app/models"
headers = {
"Content-Type" : "application/json"
}
data = {
"model_ids" : model_ids
}
response = requests .delete (url , json = data , headers = headers )
return response .json ()
# 删除单个模型
result = delete_models (["my-gpt4-model" ])
if result ['status' ] == 0 :
print ("模型删除成功" )
# 批量删除多个模型
result = delete_models (["model1" , "model2" , "model3" ])
if result ['status' ] == 0 :
print ("批量删除成功" )
# 删除单个模型
curl -X DELETE http://localhost:8088/api/v1/app/models \
-H " Content-Type: application/json" \
-d ' {
"model_ids": ["my-gpt4-model"]
}'
# 批量删除多个模型
curl -X DELETE http://localhost:8088/api/v1/app/models \
-H " Content-Type: application/json" \
-d ' {
"model_ids": ["model1", "model2", "model3"]
}'
{
"status" : 0 ,
"message" : " 删除成功" ,
"data" : null
}
除了通过API创建的数据库模型外,系统还支持通过YAML配置文件定义系统级模型。
db/model.yaml
- model_id : system_default
model_name : deepseek-chat
token : sk-your-api-key
base_url : https://api.deepseek.com/v1
note : 系统默认模型
limit : 1000
default :
- mcp_scan
- ai_infra_scan
- model_id : eval_model
model_name : gpt-4
token : sk-your-eval-key
base_url : https://api.openai.com/v1
note : 评估模型
limit : 2000
default :
- model_redteam_report
字段名
类型
必填
说明
model_id
string
是
模型ID
model_name
string
是
模型名称
token
string
是
API密钥
base_url
string
是
基础URL
note
string
否
备注信息
limit
integer
否
请求限制
default
array
否
默认使用此模型的任务类型列表
YAML配置的模型为只读 ,不支持通过API进行修改和删除
YAML配置的模型在获取列表和详情时会与数据库模型合并返回
default字段为YAML模型特有,用于标识该模型适用的默认任务类型
系统启动时自动加载YAML配置
URL : /api/v1/app/taskapi/status/{id}
方法 : GET
参数名
类型
必填
说明
id
string
是
任务会话ID
字段名
类型
说明
session_id
string
任务会话ID
status
string
任务状态:pending、running、completed、failed
title
string
任务标题
created_at
integer
创建时间戳(毫秒)
updated_at
integer
更新时间戳(毫秒)
log
string
任务执行日志
def get_task_status (session_id ):
url = f"http://localhost:8088/api/v1/app/taskapi/status/{ session_id } "
response = requests .get (url )
return response .json ()
# 使用示例
status = get_task_status ("550e8400-e29b-41d4-a716-446655440000" )
print (f"任务状态: { status ['data' ]['status' ]} " )
print (f"执行日志: { status ['data' ]['log' ]} " )
curl -X GET http://localhost:8088/api/v1/app/taskapi/status/550e8400-e29b-41d4-a716-446655440000
URL : /api/v1/app/taskapi/result/{id}
方法 : GET
参数名
类型
必填
说明
id
string
是
任务会话ID
返回详细的扫描结果,包括:
发现的漏洞列表
安全评估报告
修复建议
风险等级评估
def get_task_result (session_id ):
url = f"http://localhost:8088/api/v1/app/taskapi/result/{ session_id } "
response = requests .get (url )
return response .json ()
# 使用示例
result = get_task_result ("550e8400-e29b-41d4-a716-446655440000" )
if result ['status' ] == 0 :
print ("扫描结果:" )
print (json .dumps (result ['data' ], indent = 2 , ensure_ascii = False ))
else :
print (f"获取结果失败: { result ['message' ]} " )
curl -X GET http://localhost:8088/api/v1/app/taskapi/result/550e8400-e29b-41d4-a716-446655440000
import requests
import time
import json
def complete_mcp_scan_workflow ():
base_url = "http://localhost:8088"
# 1. 上传源码文件
print ("1. 上传源码文件..." )
upload_url = f"{ base_url } /api/v1/app/taskapi/upload"
with open ("mcp_source.zip" , 'rb' ) as f :
files = {'file' : f }
upload_response = requests .post (upload_url , files = files )
if upload_response .json ()['status' ] != 0 :
raise Exception ("文件上传失败" )
fileUrl = upload_response .json ()['data' ]['fileUrl' ]
print (f"文件上传成功: { fileUrl } " )
# 2. 创建MCP扫描任务
print ("2. 创建MCP扫描任务..." )
task_url = f"{ base_url } /api/v1/app/taskapi/tasks"
task_data = {
"type" : "mcp_scan" ,
"content" : {
"prompt" : "扫描此MCP服务器" ,
"model" : {
"model" : "gpt-4" ,
"token" : "sk-your-api-key" ,
"base_url" : "https://api.openai.com/v1"
},
"thread" : 4 ,
"language" : "zh" ,
"attachments" : fileUrl
}
}
task_response = requests .post (task_url , json = task_data )
if task_response .json ()['status' ] != 0 :
raise Exception ("任务创建失败" )
session_id = task_response .json ()['data' ]['session_id' ]
print (f"任务创建成功,会话ID: { session_id } " )
# 3. 轮询任务状态
print ("3. 监控任务执行..." )
status_url = f"{ base_url } /api/v1/app/taskapi/status/{ session_id } "
while True :
status_response = requests .get (status_url )
status_data = status_response .json ()
if status_data ['status' ] != 0 :
raise Exception ("获取任务状态失败" )
task_status = status_data ['data' ]['status' ]
print (f"当前状态: { task_status } " )
if task_status == "completed" :
print ("任务执行完成!" )
break
elif task_status == "failed" :
raise Exception ("任务执行失败" )
time .sleep (10 ) # 等待10秒后再次检查
# 4. 获取扫描结果
print ("4. 获取扫描结果..." )
result_url = f"{ base_url } /api/v1/app/taskapi/result/{ session_id } "
result_response = requests .get (result_url )
if result_response .json ()['status' ] != 0 :
raise Exception ("获取扫描结果失败" )
scan_results = result_response .json ()['data' ]
print ("扫描结果:" )
print (json .dumps (scan_results , indent = 2 , ensure_ascii = False ))
return scan_results
# 执行完整流程
if __name__ == "__main__" :
try :
results = complete_mcp_scan_workflow ()
print ("MCP扫描完成!" )
except Exception as e :
print (f"扫描失败: { e } " )
def complete_redteam_workflow ():
base_url = "http://localhost:8088"
# 1. 创建大模型安全体检
print ("1. 创建大模型安全体检任务..." )
task_url = f"{ base_url } /api/v1/app/taskapi/tasks"
task_data = {
"type" : "model_redteam_report" ,
"content" : {
"model" : [
{
"model" : "gpt-4" ,
"token" : "sk-your-api-key" ,
"base_url" : "https://api.openai.com/v1"
}
],
"eval_model" : {
"model" : "gpt-4" ,
"token" : "sk-your-eval-key" ,
"base_url" : "https://api.openai.com/v1"
},
"dataset" : {
"dataFile" : [
"JailBench-Tiny" ,
"JailbreakPrompts-Tiny" ,
"ChatGPT-Jailbreak-Prompts"
],
"numPrompts" : 100 ,
"randomSeed" : 42
}
}
}
task_response = requests .post (task_url , json = task_data )
if task_response .json ()['status' ] != 0 :
raise Exception ("任务创建失败" )
session_id = task_response .json ()['data' ]['session_id' ]
print (f"大模型安全体检任务创建成功,会话ID: { session_id } " )
# 2. 监控任务执行
print ("2. 监控任务执行..." )
status_url = f"{ base_url } /api/v1/app/taskapi/status/{ session_id } "
while True :
status_response = requests .get (status_url )
status_data = status_response .json ()
if status_data ['status' ] != 0 :
raise Exception ("获取任务状态失败" )
task_status = status_data ['data' ]['status' ]
print (f"当前状态: { task_status } " )
if task_status == "completed" :
print ("红队测评完成!" )
break
elif task_status == "failed" :
raise Exception ("大模型安全体检失败" )
time .sleep (30 ) # 大模型安全体检通常需要更长时间
# 3. 获取测评结果
print ("3. 获取测评结果..." )
result_url = f"{ base_url } /api/v1/app/taskapi/result/{ session_id } "
result_response = requests .get (result_url )
if result_response .json ()['status' ] != 0 :
raise Exception ("获取测评结果失败" )
redteam_results = result_response .json ()['data' ]
print ("大模型安全体检结果:" )
print (json .dumps (redteam_results , indent = 2 , ensure_ascii = False ))
return redteam_results
# 执行红队测评流程
if __name__ == "__main__" :
try :
results = complete_redteam_workflow ()
print ("大模型安全体检完成!" )
except Exception as e :
print (f"大模型安全体检失败: { e } " )
状态码
说明
解决方案
0
成功
-
1
失败
查看message字段获取详细错误信息
def handle_api_response (response ):
"""处理API响应的通用函数"""
data = response .json ()
if data ['status' ] == 0 :
return data ['data' ]
else :
raise Exception (f"API调用失败: { data ['message' ]} " )
# 使用示例
try :
result = handle_api_response (response )
print ("操作成功:" , result )
except Exception as e :
print ("操作失败:" , str (e ))
认证 : 确保在请求头中包含正确的认证信息
文件大小 : 上传文件大小限制请参考服务器配置
超时设置 : 根据任务复杂度合理设置超时时间
并发限制 : 避免同时创建过多任务,以免影响系统性能
结果保存 : 及时保存扫描结果,避免数据丢失
数据集选择 : 根据测试需求选择合适的数据集组合
模型配置 : 确保测试模型和评估模型配置正确
模型ID唯一性 : 创建模型时,model_id必须全局唯一
Token安全 : API密钥在返回时会自动脱敏显示为********,前端显示和编辑时需要注意
Token更新 : 更新模型时,如果token字段为空或********,则不会更新token,保持原值
模型验证 : 创建模型时系统会自动验证token和base_url的有效性
YAML模型 : 通过YAML配置的模型为只读,不支持通过API修改或删除
批量删除 : 删除模型时支持传入多个model_id进行批量删除
权限控制 : 只有模型的创建者才能查看、修改和删除该模型
如有问题,请联系技术支持团队或查看项目文档。