宝软数字 · 实操教程·进阶玩法 · 2025-11-17
截至目前，我们的教程都在讲述「在EIOS平台内部」怎么使用各项功能。但还有一类使用场景：你已经有了一套自己的系统——可能是自研的ERP、可能是钉钉上的自定义应用、可能是一个数据看板——你想把EIOS的AI能力嵌入到你现有的系统里，而不是让用户跑到EIOS平台上来操作。这时你就需要API。
EIOS的RESTful API按功能域组织为六大模块，覆盖了平台的全部核心能力：
数据API：数据集的创建、查询、更新、删除。数据的导入（支持文件上传和流式推送）、数据的查询（支持条件过滤、聚合、分页）、数据集的导出。API地址前缀：https://api.isoftbao.com/v1/data。
Agent API：Agent的创建与配置、触发Agent执行任务、查询Agent的执行状态和结果、管理Agent的生命周期（启动/暂停/删除）。API地址前缀：https://api.isoftbao.com/v1/agents。
分析API：调用AI执行特定的分析任务——销售预测、异常检测、趋势分析、根因分析、文本摘要。你可以直接调用分析能力，而不需要先创建一个持久化的Agent。API地址前缀：https://api.isoftbao.com/v1/analytics。
报告API：生成和导出各类报告——月度经营报告、自定义报表、图表数据。支持PDF和Excel导出格式。API地址前缀：https://api.isoftbao.com/v1/reports。
用户与权限API：管理用户账号、角色、权限。支持SSO集成和企业微信/钉钉的组织架构同步。API地址前缀：https://api.isoftbao.com/v1/users。
通知API：发送各类通知——邮件、短信、企业微信、钉钉。支持模板化消息和批量发送。API地址前缀：https://api.isoftbao.com/v1/notifications。
这篇教程不会逐一罗列所有接口（完整的API Reference文档在开发者中心 https://www.isoftbao.com/docs/api 可以查阅），而是聚焦在最实用的5个场景，给出可以直接复制使用的Python和JavaScript代码。从获取API密钥开始，到完整的错误处理结束。
API的使用前提：你需要对HTTP协议和RESTful API有基本了解，并且熟悉至少一种编程语言（Python或JavaScript）。本教程假设你具备这些基础知识。如果你完全零基础，建议先学习一下Python的requests库或JavaScript的fetch API的基础用法。
所有API调用都需要认证。EIOS使用API Key加签名的双重认证机制，确保即使API Key泄露，没有签名私钥也无法伪造请求。
第一步：生成API Key。登录EIOS → 右上角头像 → 「开发者中心」→「API管理」→ 点击「创建API Key」。填写Key的名称（便于你区分不同用途的Key，如「ERP对接」「数据看板」「自动化脚本」），选择Key的权限范围（强烈建议按最小权限原则创建——只勾选该Key实际需要的API模块）。点击创建后，系统生成一个Access Key ID（公开标识，类似用户名）和一个Access Key Secret（密钥，类似密码）。
重要警告：Access Key Secret只在创建时显示一次，请立即复制并保存到安全的地方（如密码管理器或环境变量中）。如果丢失，只能重新生成新的Key。绝对不要将Secret硬编码在源代码中提交到Git仓库！使用环境变量或密钥管理服务。
第二步：理解签名机制。每个API请求需要在HTTP Header中携带以下认证信息：X-Access-Key-Id（你的Access Key ID）、X-Timestamp（请求发出的Unix时间戳，秒级）、X-Signature（请求签名）。签名通过HMAC-SHA256算法生成。Python示例代码如下：
import hmac
import hashlib
import time
import requests
ACCESS_KEY_ID = "你的Access Key ID"
ACCESS_KEY_SECRET = "你的Access Key Secret"
def build_signature(method, path, body, timestamp):
"""构建请求签名"""
message = f"{method}\n{path}\n{body}\n{timestamp}"
signature = hmac.new(
ACCESS_KEY_SECRET.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
).hexdigest()
return signature
def api_request(method, path, body=""):
"""发送带签名的API请求"""
timestamp = str(int(time.time()))
signature = build_signature(method, path, body, timestamp)
headers = {
"X-Access-Key-Id": ACCESS_KEY_ID,
"X-Timestamp": timestamp,
"X-Signature": signature,
"Content-Type": "application/json"
}
url = f"https://api.isoftbao.com{path}"
response = requests.request(method, url, headers=headers, data=body)
return response.json()第三步：验证连接。运行一个最简单的API调用来验证你的Key配置正确。调用健康检查接口GET /v1/health，不需要参数，不需要签名。如果返回状态码200和服务器状态信息，说明网络连通性没问题。然后再调用一个需要签名的接口（如GET /v1/data/datasets）来验证签名机制正确。
签名机制的价值：即使攻击者截获了你的API请求（比如在不安全的网络中），他拿到了Access Key ID和时间戳，但没有Access Key Secret，无法伪造签名。更重要的是，时间戳机制可以防止重放攻击——每个签名在5分钟内有效，过了5分钟旧签名自动失效。这就是为什么我们要求你在每个请求的Header中都携带当前时间戳。
以下是5个最常见的API使用场景，每个都配有完整的Python代码示例。你可以直接复制到你的项目中，替换参数即可使用。
场景一：导入销售数据。你的ERP系统每天生成销售数据CSV文件，你想让这些数据自动流入EIOS供Agent分析。使用数据导入API，支持文件上传。
def import_sales_data(csv_file_path):
"""将CSV销售数据导入EIOS"""
path = "/v1/data/datasets/import"
timestamp = str(int(time.time()))
# 文件上传使用multipart/form-data格式
with open(csv_file_path, 'rb') as f:
files = {'file': ('sales_data.csv', f, 'text/csv')}
data = {
'dataset_name': 'ERP销售数据',
'domain': 'sales',
'date_column': 'order_date',
'encoding': 'utf-8'
}
# 文件上传的签名逻辑略有不同，body取JSON序列化的data部分
body_str = json.dumps(data)
signature = build_signature('POST', path, body_str, timestamp)
headers = {
"X-Access-Key-Id": ACCESS_KEY_ID,
"X-Timestamp": timestamp,
"X-Signature": signature
}
response = requests.post(
f"https://api.isoftbao.com{path}",
headers=headers,
files=files,
data=data
)
return response.json()场景二：触发Agent执行分析。你的系统在某个业务流程完成后（如月度结账完成），自动触发对应的Agent执行分析，并获取分析结果。
def trigger_agent(agent_id, params=None):
"""触发指定Agent执行任务"""
path = "/v1/agents/execute"
body = json.dumps({
"agent_id": agent_id,
"params": params or {},
"async_mode": False # False=同步等待结果, True=异步立即返回任务ID
})
result = api_request('POST', path, body)
if result.get('success'):
if result.get('async_mode'):
task_id = result['data']['task_id']
print(f"任务已提交，task_id: {task_id}，稍后查询结果")
return task_id
else:
analysis_result = result['data']['output']
print("分析完成！")
print(analysis_result)
return analysis_result
else:
print(f"执行失败: {result.get('error', {}).get('message')}")
def query_task_result(task_id):
"""查询异步任务的结果"""
path = f"/v1/agents/tasks/{task_id}"
result = api_request('GET', path)
status = result['data']['status']
if status == 'completed':
return result['data']['output']
elif status == 'failed':
print(f"任务失败: {result['data']['error']}")
return None
else:
print(f"任务状态: {status}，请稍后再查询")
return None场景三：查询数据集并进行聚合分析。你的数据看板需要展示本月各产品线的销售额汇总。直接通过API查询并聚合数据，不经过EIOS的前端页面。
def query_monthly_sales_by_product(date_from, date_to):
"""查询指定日期范围内的销售额，按产品线聚合"""
path = "/v1/data/query"
body = json.dumps({
"dataset_id": "ds_sales_orders",
"filters": [
{"field": "order_date", "operator": "between", "value": [date_from, date_to]}
],
"group_by": ["product_line"],
"aggregates": [
{"field": "amount", "function": "sum", "alias": "total_revenue"},
{"field": "quantity", "function": "sum", "alias": "total_quantity"},
{"field": "order_id", "function": "count", "alias": "order_count"}
],
"order_by": [{"field": "total_revenue", "direction": "desc"}],
"limit": 20
})
result = api_request('POST', path, body)
if result.get('success'):
rows = result['data']['rows']
for row in rows:
product = row['product_line']
revenue = row['total_revenue']
print(f"{product}: ¥{revenue:,.2f}")
return rows
return []场景四：生成月度经营报告并导出PDF。每月自动生成经营分析报告，导出为PDF文件保存到指定路径。
def generate_monthly_report(year, month):
"""生成指定月份的月度经营分析报告PDF"""
# 1. 提交报告生成请求
path = "/v1/reports/generate"
body = json.dumps({
"template_id": "tpl_monthly_business",
"params": {
"year": year,
"month": month
},
"format": "pdf"
})
result = api_request('POST', path, body)
if not result.get('success'):
print(f"报告生成请求失败: {result.get('error')}")
return None
report_id = result['data']['report_id']
print(f"报告生成任务已提交，report_id: {report_id}")
# 2. 轮询等待报告生成完成
import time as time_mod
for _ in range(30): # 最多等待30轮
time_mod.sleep(2)
path = f"/v1/reports/{report_id}"
status_result = api_request('GET', path)
if status_result['data']['status'] == 'completed':
download_url = status_result['data']['download_url']
# 3. 下载PDF文件
# 注意：下载URL不需要API签名
pdf_response = requests.get(download_url)
filename = f"monthly_report_{year}_{month:02d}.pdf"
with open(filename, 'wb') as f:
f.write(pdf_response.content)
print(f"报告已下载: {filename}")
return filename
elif status_result['data']['status'] == 'failed':
print("报告生成失败")
return None
print("报告生成超时")
return None场景五：批量查询多个Agent的执行状态。你有多个Agent在同时运行，需要在管理后台展示它们的状态。批量查询API可以一次获取所有Agent的最新状态。
def get_agents_status(agent_ids):
"""批量查询多个Agent的当前状态"""
path = "/v1/agents/status/batch"
body = json.dumps({
"agent_ids": agent_ids
})
result = api_request('POST', path, body)
if result.get('success'):
status_map = {}
for item in result['data']['agents']:
agent_id = item['agent_id']
status = item['status']
last_run = item.get('last_run_at', '从未运行')
task_count = item.get('pending_tasks', 0)
status_map[agent_id] = {
'status': status,
'last_run': last_run,
'pending_tasks': task_count
}
print(f"Agent {agent_id}: {status} | 上次运行: {last_run} | 待处理: {task_count}")
return status_map
return {}
代码组织建议：把签名逻辑封装成一个独立的模块（如eios_client.py），在你的项目各处引用。不要把Access Key Secret散落在各个文件中——一个统一的客户端模块更容易管理和审计。最好把这个客户端模块打包成一个内部Python包，通过pip安装到各个项目中。
如果你的系统是Web应用，你可能需要在浏览器端或Node.js后端调用EIOS API。以下是两个最常用的JavaScript集成场景。
场景一：Node.js后端调用EIOS API（推荐方案）。生产环境中强烈建议API调用在后端进行，避免Access Key Secret暴露在前端代码中。使用Express框架作为中间层，前端请求你的后端，你的后端请求EIOS API。
// eios-client.js - Node.js EIOS API客户端
const crypto = require('crypto');
const https = require('https');
const ACCESS_KEY_ID = process.env.EIOS_ACCESS_KEY_ID;
const ACCESS_KEY_SECRET = process.env.EIOS_ACCESS_KEY_SECRET;
const API_BASE = 'api.isoftbao.com';
function buildSignature(method, path, body, timestamp) {
const message = `${method}\n${path}\n${body}\n${timestamp}`;
return crypto
.createHmac('sha256', ACCESS_KEY_SECRET)
.update(message)
.digest('hex');
}
async function apiRequest(method, path, body = '') {
const timestamp = Math.floor(Date.now() / 1000).toString();
const signature = buildSignature(method, path, body, timestamp);
return new Promise((resolve, reject) => {
const options = {
hostname: API_BASE,
path: path,
method: method,
headers: {
'X-Access-Key-Id': ACCESS_KEY_ID,
'X-Timestamp': timestamp,
'X-Signature': signature,
'Content-Type': 'application/json'
}
};
const req = https.request(options, (res) => {
let data = '';
res.on('data', chunk => data += chunk);
res.on('end', () => resolve(JSON.parse(data)));
});
req.on('error', reject);
if (body) req.write(body);
req.end();
});
}
// Express路由示例：前端通过你的后端查询数据
app.get('/api/eiose/sales-summary', async (req, res) => {
try {
const result = await apiRequest('POST', '/api/v1/data/query', JSON.stringify({
dataset_id: 'ds_sales',
group_by: ['product_category'],
aggregates: [{ field: 'amount', function: 'sum', alias: 'total' }]
}));
res.json(result);
} catch (error) {
res.status(500).json({ error: error.message });
}
});场景二：前端直接调用（仅适用于低安全场景）。如果你需要在浏览器中直接调用EIOS API（比如一个内部管理工具），请务必使用临时令牌而非永久API Key。临时令牌由后端生成，有效期短（默认15分钟），只能用于特定操作。
// 前端JavaScript - 使用临时Token调用EIOS API
async function fetchWithToken(path, options = {}) {
// 1. 先从你的后端获取临时Token
const tokenResponse = await fetch('/api/eiose/token', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
scope: options.scope || 'read',
resource: options.resource || ''
})
});
const { token } = await tokenResponse.json();
// 2. 使用临时Token调用EIOS API
const response = await fetch(`https://api.isoftbao.com${path}`, {
...options,
headers: {
...options.headers,
'Authorization': `Bearer ${token}`,
'Content-Type': 'application/json'
}
});
return response.json();
}
// 使用示例：获取某Agent的分析报告
async function getAnalysisReport(agentId) {
const report = await fetchWithToken(`/api/v1/agents/${agentId}/reports/latest`, {
scope: 'read',
resource: `agent:${agentId}`
});
document.getElementById('report-container').innerHTML =
report.data.summary;
}
前端安全的底线：永远不要把Access Key Secret放在前端代码中。即使你用了代码混淆、环境变量、构建时替换——浏览器的运行环境本质上是不安全的，任何在前端出现的密钥都可能被提取。唯一的正确做法是通过后端生成临时令牌。如果你的业务场景无法满足这个要求（比如纯静态网站没有后端），那么至少使用一个Cloudflare Worker或Vercel Edge Function作为轻量级的中间代理层。
在开发和测试环境中，API调用几乎总是成功的。但在生产环境中，网络抖动、服务器限流、数据量过大、并发冲突等问题会频繁出现。一套健壮的错误处理和重试机制是生产级集成的必备要素。
EIOS API的错误码体系：所有API响应都包含统一的错误结构。HTTP状态码400表示客户端错误（参数错误、格式错误），401表示认证失败（Key过期或签名错误），403表示权限不足（Key没有该操作的权限），429表示请求过于频繁（触发了限流），500表示服务器内部错误。响应体中的error.code字段提供了更细粒度的错误码，如DATA_NOT_FOUND（数据不存在）、AGENT_BUSY（Agent正在执行其他任务）、RATE_LIMIT_EXCEEDED（超出调用频率限制）。
指数退避重试策略：对于可重试的错误（网络超时、服务器500、429限流），使用指数退避策略。不可重试的错误（401认证失败、403权限不足、400参数错误）应该立即失败并记录日志。Python实现：
import random
import time as time_mod
RETRYABLE_ERRORS = {'NETWORK_TIMEOUT', 'SERVER_ERROR', 'RATE_LIMIT_EXCEEDED'}
MAX_RETRIES = 3
BASE_DELAY = 1.0 # 基础延迟秒数
def api_request_with_retry(method, path, body=""):
"""带指数退避重试的API请求"""
for attempt in range(MAX_RETRIES + 1):
try:
result = api_request(method, path, body)
if result.get('success'):
return result
error_code = result.get('error', {}).get('code', '')
# 不可重试错误，直接返回
if error_code not in RETRYABLE_ERRORS:
print(f"不可重试错误 [{error_code}]，放弃")
return result
# 可重试错误，判断是否还有重试次数
if attempt < MAX_RETRIES:
# 指数退避: 1s, 2s, 4s + 随机抖动
delay = BASE_DELAY * (2 ** attempt) + random.uniform(0, 0.5)
print(f"遇到可重试错误 [{error_code}]，{delay:.1f}秒后重试 (第{attempt+1}次)")
time_mod.sleep(delay)
else:
print(f"重试{MAX_RETRIES}次后仍失败")
return result
except requests.exceptions.Timeout:
if attempt < MAX_RETRIES:
delay = BASE_DELAY * (2 ** attempt)
print(f"请求超时，{delay:.1f}秒后重试")
time_mod.sleep(delay)
else:
raise # 重试耗尽，向上抛出异常限流处理：EIOS API有速率限制。默认情况下，每个API Key每秒钟最多调用10次，每分钟最多调用200次。当你收到429状态码时，响应Header中的Retry-After字段（单位秒）指示了你需要等待多久才能重试。在重试逻辑中优先使用这个值：
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 5))
print(f"触发限流，等待{retry_after}秒")
time_mod.sleep(retry_after)
# 使用更长的退避间隔重试批量操作的并发控制：当你需要调用大量API（如批量导入100个文件），不要无脑并发——那样会立即触发限流。使用信号量控制并发数，建议并发数不超过5。
import asyncio
async def batch_import_with_concurrency(file_paths):
"""批量导入文件，并发控制"""
semaphore = asyncio.Semaphore(5) # 最多5个并发
async def import_one(file_path):
async with semaphore:
# 在异步环境中调用同步的导入函数
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None, import_sales_data, file_path
)
return result
tasks = [import_one(fp) for fp in file_paths]
results = await asyncio.gather(*tasks, return_exceptions=True)
success_count = sum(1 for r in results if isinstance(r, dict) and r.get('success'))
print(f"批量导入完成: {success_count}/{len(file_paths)} 成功")
return results
生产环境的日志是不可或缺的：每一次API调用都记录下：调用时间、API路径、响应状态码、耗时（毫秒）、错误信息（如有）。这些日志在排查问题时是无价之宝——你能精确地回答「上个月销售数据导入失败了多少次？原因都是什么？」而不是「好像有时候失败但我也不知道为什么」。使用结构化日志（JSON格式），方便后续接入日志分析平台。
在前面的五个场景之外，还有两个高级用法值得单独说明。
Webhook集成（将在下一篇教程详解）：前面的所有场景都是「你的系统主动调用EIOS」。但有时候你需要反过来——当EIOS中的某些事件发生时，EIOS主动通知你的系统。这就是Webhook。例如：当Agent完成一项分析任务，EIOS通过Webhook将结果推送到你的系统；当库存Agent检测到某SKU库存低于安全值，自动推送预警到你的ERP或钉钉群。Webhook让你的系统和EIOS之间从「单向调用」升级为「双向通信」。
批量数据推送（Data Push）：如果你的系统每天产生大量数据（如上万条销售记录），使用单条API逐条推送效率太低。使用批量推送接口，一次推送最多1000条记录，系统会在后台异步处理。API路径：POST /v1/data/datasets/{dataset_id}/records/batch。
使用SDK简化开发：如果你不想从头实现签名和重试逻辑，EIOS提供了官方Python SDK（pip install eios-sdk）和Node.js SDK（npm install @baodoo/eios-sdk）。SDK封装了认证、签名、重试、并发控制等所有底层逻辑，你只需要关注业务逻辑。完整文档在 https://github.com/baodoo/eios-sdk。
选择调用方式的决策树：如果你的系统需要「定时拉取EIOS数据」（如每天凌晨同步一次销售数据），用API主动调用。如果你的系统需要「实时响应EIOS事件」（如Agent一完成分析就通知），用Webhook。如果你的系统需要「高频、大批量写入数据」，用批量推送接口加并发控制。如果你的团队没有精力处理底层API细节，用官方SDK。不同的需求有对应的最优方式，不要用一种方式解决所有问题。
API是EIOS平台开放能力的核心通道。通过API，你可以把EIOS的AI能力无缝嵌入到已有的业务系统中，而不是让用户在不同的平台间来回切换。下一篇（也是进阶玩法的收官教程）我们将专门讲解Webhook配置——如何让EIOS主动推送到你的系统，实现真正的双向智能联动。