部署到 VeFaaS

veadk init

$ veadk init
Welcome use VeADK to create your project. We will generate a `weather-reporter` application for you.
Local directory name [veadk-cloud-proj]:
Volcengine FaaS application name [veadk-cloud-agent]:
Volcengine API Gateway instance name []:
Volcengine API Gateway service name []:
Volcengine API Gateway upstream name []:
Choose a deploy mode:
  1. A2A/MCP Server
  2. VeADK Web / Google ADK Web
Enter your choice (1, 2): 1
Choose an authentication method:
  1. None
  2. API key
  3. OAuth2
Enter your choice (1, 2, 3): 1
Template project has been generated at .../veadk-cloud-proj
Edit .../veadk-cloud-proj/src to define your agents
Edit .../veadk-cloud-proj/deploy.py to define your deployment attributes
Run python `deploy.py` for deployment on Volcengine FaaS platform.

volcengine:
  access_key: "YOUR_ACCESS_KEY"    # 替换为你的火山引擎 Access Key
  secret_key: "YOUR_SECRET_KEY"  # 替换为你的火山引擎 Secret Key

python deploy.py

VeFaaS application ID: 3b015fxxxxxx
Test message: How is the weather like in Beijing?

veadk deploy --vefaas-app-name {YourAgentAppName}

from veadk.cloud.cloud_agent_engine import CloudAgentEngine

# 1. 初始化云引擎，凭证将从环境变量自动获取
engine = CloudAgentEngine()

# 2. 调用 deploy 方法部署应用
cloud_app = engine.deploy(
    application_name="my-python-agent",
    path="./path/to/your/agent/project",
    gateway_name="my-agent-gateway",
    use_adk_web=False
)

# 部署成功后，可以获取应用的访问端点
print(f"Application endpoint: {cloud_app.vefaas_endpoint}")

import asyncio
from veadk.cloud.cloud_app import CloudApp

# 通过部署时返回的应用名称构造 CloudApp（也可传入 vefaas_endpoint 或 vefaas_application_id）
cloud_app = CloudApp(vefaas_application_name="my-python-agent")

SESSION_ID = "test_session_123"
USER_ID = "test_user_001"

async def main():
    response = await cloud_app.message_send(
        "请问北京今天天气怎么样？", SESSION_ID, USER_ID
    )
    print(f"Agent response: {response}")

asyncio.run(main())

import asyncio
from veadk.cloud.cloud_app import CloudApp

APP_ENDPOINT = "<YOUR_APPLICATION_ENDPOINT>"
SESSION_ID = "test_session"
USER_ID = "test_user"

app = CloudApp(vefaas_endpoint=APP_ENDPOINT)

# 调用方式相同：message, session_id, user_id
response = asyncio.run(app.message_send("深圳呢？", SESSION_ID, USER_ID))
print(f"Agent response: {response}")

from veadk.cloud.cloud_agent_engine import CloudAgentEngine

engine = CloudAgentEngine()

# 更新现有应用的代码，保持相同的访问端点
updated_cloud_app = engine.update_function_code(
    application_name="my-agent-app",  # 现有应用名称
    path="/my-agent-project"        # 本地项目路径，确保项目路径包含agent.py文件
)

# 可以使用updated_cloud_app.vefaas_endpoint访问您的项目

from veadk.cloud.cloud_agent_engine import CloudAgentEngine

engine = CloudAgentEngine()

# 调用 remove 方法删除指定应用
engine.remove(app_name="my-python-agent")

Confirm delete cloud app my-python-agent? (y/N): y

import asyncio
from veadk.cloud.cloud_app import CloudApp

# -- 配置 --
ENDPOINT = "<YOUR_APPLICATION_ENDPOINT>"  # 替换为你的应用访问端点
SESSION_ID = "test_session"
USER_ID = "test_user"
TEST_MESSAGE = "北京今天的天气怎么样？"

async def main():
    # 使用端点初始化 CloudApp 实例
    app = CloudApp(vefaas_endpoint=ENDPOINT)

    # 发送消息
    response = await app.message_send(TEST_MESSAGE, SESSION_ID, USER_ID)
    print(f"A2A Response: {response}")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from veadk import Agent, Runner
from veadk.a2a.remote_ve_agent import RemoteVeAgent

# -- 配置 --
ENDPOINT = "<YOUR_APPLICATION_ENDPOINT>"      # 替换为你的应用访问端点
KEY_AUTH_TOKEN = "<YOUR_KEY_AUTH_TOKEN>"  # 替换为你的 Key Auth Token
SESSION_ID = "test_session"
USER_ID = "test_user"
TEST_MESSAGE = "上海呢？"

async def main():
    # 1. 将云端 Agent 封装为 RemoteVeAgent
    remote_agent = RemoteVeAgent(
        name="cloud_weather_agent",
        url=ENDPOINT,
        auth_token=KEY_AUTH_TOKEN,
        auth_method="querystring"  # 指定 token 通过查询字符串传递
    )

    # 2. 创建一个本地 Agent，并将远程 Agent 作为其子 Agent
    local_agent = Agent(
        name="local_query_agent",
        instruction="将用户问题转发给云端 Agent 并返回结果。",
        sub_agents=[remote_agent],
    )

    # 3. 使用 Runner 执行任务
    runner = Runner(agent=local_agent, user_id=USER_ID)
    response = await runner.run(TEST_MESSAGE, session_id=SESSION_ID)
    print(f"A2A Remote Response: {response}")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from fastmcp import Client

# -- 配置 --
ENDPOINT = "<YOUR_APPLICATION_ENDPOINT>"      # 替换为你的应用访问端点
KEY_AUTH_TOKEN = "<YOUR_KEY_AUTH_TOKEN>"  # 替换为你的 Key Auth Token
SESSION_ID = "my_test_session_03"
USER_ID = "my_test_user_03"
TEST_MESSAGE = "广州今天天气如何？"

async def main():
    # 构造带有认证 token 的 MCP 端点 URL
    mcp_endpoint = f"{ENDPOINT}/mcp?token={KEY_AUTH_TOKEN}"

    # 如果关闭了网关的密钥认证，则无需 token 参数
    # mcp_endpoint = f"{ENDPOINT}/mcp"

    client = Client(mcp_endpoint)
    async with client:
        # 发现可用的工具 (Agent)
        tools = await client.list_tools()
        print(f"MCP Available Tools: {tools}")

        # 调用 run_agent 工具
        response = await client.call_tool(
            "run_agent",
            {
                "user_input": TEST_MESSAGE,
                "session_id": SESSION_ID,
                "user_id": USER_ID,
            },
        )
        print(f"MCP Response: {response}")

if __name__ == "__main__":
    asyncio.run(main())

veadk pipeline --github-url <YOUR_GITHUB_REPO_URL> --github-branch main --github-token <YOUR_GITHUB_TOKEN>