文档备案控制台登录立即注册
首页文章问答视频活动下载资源开发者计划团队号镜像站
发布

火山图像服务接入扣子插件

扣子售后小助手
扣子售后小助手
扣子
扣子扣子专业版

创建并配置插件

详细创建流程可参考文档使用 IDE 创建插件，此处仅提供简化操作流程

  1. 创建插件与工具

  1. 创建插件

picture.image

picture.image

picture.image

  1. 创建工具

插件中的每个工具代表一个API接口，一般“查询”、“执行”等动作分别代表一个工具

picture.image

  1. 配置写入代码

  1. 添加示例代码

在这个位置添加示例代码

picture.image

示例****代码来自“接入说明”部分。服务文档中有如截图所示的接入说明部分的服务，都可以按当前方案接入

picture.image

代码链接https://www.volcengine.com/docs/6444/1390583

示例代码：

import json
import sys
import os
import base64
import datetime
import hashlib
import hmac
import requests
method = 'POST'
host = 'visual.volcengineapi.com'
region = 'cn-north-1'
endpoint = 'https://visual.volcengineapi.com'
service = 'cv'
def sign(key, msg):
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
def getSignatureKey(key, dateStamp, regionName, serviceName):
    kDate = sign(key.encode('utf-8'), dateStamp)
    kRegion = sign(kDate, regionName)
    kService = sign(kRegion, serviceName)
    kSigning = sign(kService, 'request')
    return kSigning
def formatQuery(parameters):
    request_parameters_init = ''
    for key in sorted(parameters):
        request_parameters_init += key + '=' + parameters[key] + '&'
    request_parameters = request_parameters_init[:-1]
    return request_parameters
def signV4Request(access_key, secret_key, service, req_query, req_body):
    if access_key is None or secret_key is None:
        print('No access key is available.')
        sys.exit()
    t = datetime.datetime.utcnow()
    current_date = t.strftime('%Y%m%dT%H%M%SZ')
    # current_date = '20210818T095729Z'
    datestamp = t.strftime('%Y%m%d')  # Date w/o time, used in credential scope
    canonical_uri = '/'
    canonical_querystring = req_query
    signed_headers = 'content-type;host;x-content-sha256;x-date'
    payload_hash = hashlib.sha256(req_body.encode('utf-8')).hexdigest()
    content_type = 'application/json'
    canonical_headers = 'content-type:' + content_type + '\n' + 'host:' + host + \
        '\n' + 'x-content-sha256:' + payload_hash + \
        '\n' + 'x-date:' + current_date + '\n'
    canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + \
        '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
    # print(canonical_request)
    algorithm = 'HMAC-SHA256'
    credential_scope = datestamp + '/' + region + '/' + service + '/' + 'request'
    string_to_sign = algorithm + '\n' + current_date + '\n' + credential_scope + '\n' + hashlib.sha256(
        canonical_request.encode('utf-8')).hexdigest()
    # print(string_to_sign)
    signing_key = getSignatureKey(secret_key, datestamp, region, service)
    # print(signing_key)
    signature = hmac.new(signing_key, (string_to_sign).encode(
        'utf-8'), hashlib.sha256).hexdigest()
    # print(signature)
    authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + \
        credential_scope + ', ' + 'SignedHeaders=' + \
        signed_headers + ', ' + 'Signature=' + signature
    # print(authorization_header)
    headers = {'X-Date': current_date,
               'Authorization': authorization_header,
               'X-Content-Sha256': payload_hash,
               'Content-Type': content_type
               }
    # print(headers)
    # ************* SEND THE REQUEST *************
    request_url = endpoint + '?' + canonical_querystring
    print('\nBEGIN REQUEST++++++++++++++++++++++++++++++++++++')
    print('Request URL = ' + request_url)
    try:
        r = requests.post(request_url, headers=headers, data=req_body)
    except Exception as err:
        print(f'error occurred: {err}')
        raise
    else:
        print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
        print(f'Response code: {r.status_code}\n')
        # 使用 replace 方法将 \u0026 替换为 &
        resp_str = r.text.replace("\u0026", "&")
        print(f'Response body: {resp_str}\n')
if __name__ == "__main__":
    # 请求凭证，从访问控制申请
    access_key = 'AK*****'
    secret_key = '*****=='
    # 请求Query，按照接口文档中填入即可
    query_params = {
        'Action': 'CVProcess',
        'Version': '2022-08-31',
    }
    formatted_query = formatQuery(query_params)
    # 请求Body，按照接口文档中填入即可
    body_params = {
        "req_key": "******",
        # ......
    }
    formatted_body = json.dumps(body_params)
    
    signV4Request(access_key, secret_key, service,
                  formatted_query, formatted_body)

  1. 调整代码

调整前

picture.image

调整后

picture.image

  1. 安装requests依赖包

picture.image

  1. 配置参数输入输出

插件在使用时，如果想要每次调用动态传入不同参数作为API接口的入参，需要配置对应元数据使用。

  1. 根据具体接口文档配置想要传入的参数值。

picture.image

  1. 代码中对参数值进行变量定义

picture.image

至此，主要部分修改完毕，可以参考对应服务文档接入相关服务

示例：

示例参数根据发文时文档填写，实际使用时还请参考对应的最新接口文档

示例一：接入即梦4.0

接入即梦4.0，参考文档：https://www.volcengine.com/docs/85621/1817045

即梦4.0接口为异步接口，需要先再按上述步骤创建一个工具，作为查询接口

picture.image

按文档配置请求参数

  1. 提交任务接口

根据文档中的内容，在代码配置相关参数

picture.image

picture.image

access_key secret_key 获取

https://console.volcengine.com/iam/keymanage这个网页获取

picture.image

配置好后，调用测试，调用成功

picture.image

  1. 查询任务接口

根据文档中的内容，在代码配置相关参数。

picture.image

参数配置

picture.image

access_key secret_key 使用与提交任务相同的access_key secret_key

运行结果

picture.image

测试调用成功

注意：

1、当前测试只填写了必填参数，其他参数根据实际需求填写

2、自定义插件输出长度存在限制，不建议使用base64等超长内容输出

示例二：接入单图视频驱动

单图视频驱动产品，参考文档https://www.volcengine.com/docs/86081/1804524

按创建并配置插件部分内容，先创建两个基本插件，一个创建任务，一个查询任务

picture.image

通过https://console.volcengine.com/iam/keymanage获取access_key、secret_key

picture.image

根据文档内容填写参数

  1. 任务提交接口

设置需要传参的元数据和参数，需参考上文配置参数输入输出进行设置。

picture.image

picture.image

填写query_params、body_params和两个key

picture.image

完整Demo：

from runtime import Args
from typings.one_pic.one_pic import Input, Output

"""
Each file needs to export a function named `handler`. This function is the entrance to the Tool.

Parameters:
args: parameters of the entry function.
args.input - input parameters, you can get test input value by args.input.xxx.
args.logger - logger instance used to print logs, injected by runtime.

Remember to fill in input/output in Metadata, it helps LLM to recognize and use tool.

Return:
The return data of the function, which should match the declared output parameters.
"""
import json
import sys
import os
import base64
import datetime
import hashlib
import hmac
import requests
method = 'POST'
host = 'visual.volcengineapi.com'
region = 'cn-north-1'
endpoint = 'https://visual.volcengineapi.com'
service = 'cv'
def sign(key, msg):
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
def getSignatureKey(key, dateStamp, regionName, serviceName):
    kDate = sign(key.encode('utf-8'), dateStamp)
    kRegion = sign(kDate, regionName)
    kService = sign(kRegion, serviceName)
    kSigning = sign(kService, 'request')
    return kSigning
def formatQuery(parameters):
    request_parameters_init = ''
    for key in sorted(parameters):
        request_parameters_init += key + '=' + parameters[key] + '&'
    request_parameters = request_parameters_init[:-1]
    return request_parameters
def signV4Request(access_key, secret_key, service, req_query, req_body):
    if access_key is None or secret_key is None:
        print('No access key is available.')
        sys.exit()
    t = datetime.datetime.utcnow()
    current_date = t.strftime('%Y%m%dT%H%M%SZ')
    # current_date = '20210818T095729Z'
    datestamp = t.strftime('%Y%m%d')  # Date w/o time, used in credential scope
    canonical_uri = '/'
    canonical_querystring = req_query
    signed_headers = 'content-type;host;x-content-sha256;x-date'
    payload_hash = hashlib.sha256(req_body.encode('utf-8')).hexdigest()
    content_type = 'application/json'
    canonical_headers = 'content-type:' + content_type + '\n' + 'host:' + host + \
        '\n' + 'x-content-sha256:' + payload_hash + \
        '\n' + 'x-date:' + current_date + '\n'
    canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + \
        '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
    # print(canonical_request)
    algorithm = 'HMAC-SHA256'
    credential_scope = datestamp + '/' + region + '/' + service + '/' + 'request'
    string_to_sign = algorithm + '\n' + current_date + '\n' + credential_scope + '\n' + hashlib.sha256(
        canonical_request.encode('utf-8')).hexdigest()
    # print(string_to_sign)
    signing_key = getSignatureKey(secret_key, datestamp, region, service)
    # print(signing_key)
    signature = hmac.new(signing_key, (string_to_sign).encode(
        'utf-8'), hashlib.sha256).hexdigest()
    # print(signature)
    authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + \
        credential_scope + ', ' + 'SignedHeaders=' + \
        signed_headers + ', ' + 'Signature=' + signature
    # print(authorization_header)
    headers = {'X-Date': current_date,
               'Authorization': authorization_header,
               'X-Content-Sha256': payload_hash,
               'Content-Type': content_type
               }
    # print(headers)
    # ************* SEND THE REQUEST *************
    request_url = endpoint + '?' + canonical_querystring
    print('\nBEGIN REQUEST++++++++++++++++++++++++++++++++++++')
    print('Request URL = ' + request_url)
    try:
        r = requests.post(request_url, headers=headers, data=req_body)
    except Exception as err:
        print(f'error occurred: {err}')
        raise
    else:
        print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
        print(f'Response code: {r.status_code}\n')
        # 使用 replace 方法将 \u0026 替换为 &
        resp_str = r.text.replace("\u0026", "&")
        print(f'Response body: {resp_str}\n')
        return resp_str

def handler(args: Args[Input])->Output:
    # 引入参数
    image_url = args.input.image_url
    video_url = args.input.video_url

    # 请求凭证，从访问控制申请
    access_key = 'AKL*****************'
    secret_key = 'TTJ*********************=='
    # 请求Query，按照接口文档中填入即可
    query_params = {
        'Action': 'CVSubmitTask',
        'Version': '2022-08-31',
    }
    formatted_query = formatQuery(query_params)
    
    # 请求Body，按照接口文档中填入即可
    body_params = {
        "req_key": "realman_avatar_imitator_v2v_gen_video",
        "image_url": image_url,
        "driving_video_info":  {
            "store_type": 0,
            "video_url": video_url
        }
    }
    formatted_body = json.dumps(body_params)
    
    result = signV4Request(access_key, secret_key, service,
                  formatted_query, formatted_body)
    return {"message": result}

  1. 查询接口

  1. 设置需要传参的元数据和参数，参考上文配置参数输入输出进行设置。

picture.image

  1. 填写query_params、body_params和两个key

picture.image

{
    "req_key": "realman_avatar_imitator_v2v_gen_video",
    "task_id": "<任务提交接口返回task_id>"
}

task_id参数是由任务提交接口返回的，填写到这个部分用于查询对应任务。

注意：

1、该接口服务处理需要一点时间，发起请求后立即查询可能无法获取到结果，当 "status" 为 "done" ，表明任务处理完成，可获取对应url。

0
0
0
0
关于作者
扣子售后小助手
扣子售后小助手
关于作者
扣子售后小助手
扣子售后小助手

文章

0

获赞

0

收藏

0

相关资源
边缘云原生操作系统设计与思考
《火山引擎边缘云原生操作系统设计与思考》 徐广治 | 火山引擎边缘云资深架构师
点击下载 
相关产品
推荐阅读
扣子，正式拥抱开源！
扣子 logid 查询介绍
Coze 智能体编排介绍与常见问题排查
扣子社区文档使用指南
节点报错：The input data is incorrect as fields.....
评论
未登录
看完啦，登录分享一下感受吧～
暂无评论