出站认证

from veadk.integrations.ve_identity import VeIdentityFunctionTool, VeIdentityMcpToolset
from google.adk.agents.mcp import StdioServerParameters

# 函数工具：auth_config 决定凭据来源，into 指定注入形参
tool = VeIdentityFunctionTool(func=call_api, auth_config=auth_config, into="api_key")

# MCP 工具集：同样传入 auth_config
toolset = VeIdentityMcpToolset(
    auth_config=auth_config,
    connection_params=StdioServerParameters(command="python", args=["-m", "my_mcp_server"]),
)

from veadk.integrations.ve_identity import api_key_auth

auth_config = api_key_auth(provider_name="my-api-provider")

from veadk.integrations.ve_identity import oauth2_auth

auth_config = oauth2_auth(
    provider_name="my-oauth2-m2m-provider",
    scopes=["api://your-service/.default"],
    auth_flow="M2M",
)

from veadk.integrations.ve_identity import oauth2_auth

auth_config = oauth2_auth(
    provider_name="github-oauth2-provider",
    scopes=["repo", "user"],
    auth_flow="USER_FEDERATION",
    callback_url="https://your-app.com/oauth/callback",
)

import asyncio
from veadk import Agent
from veadk.integrations.ve_identity import VeIdentityMcpToolset, oauth2_auth
from veadk.integrations.ve_identity.auth_processor import AuthRequestProcessor
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams

ecs_tools = VeIdentityMcpToolset(
    auth_config=oauth2_auth(
        provider_name="volc-ecs-oauth2-provider",
        scopes=["read"],
        auth_flow="USER_FEDERATION",
    ),
    connection_params=StreamableHTTPConnectionParams(url="https://ecs.mcp.volcbiz.com/ecs/mcp"),
)

agent = Agent(
    tools=[ecs_tools],
    system_prompt="你是火山引擎 ECS 助手，可以查询 ECS 实例信息并执行服务器命令。",
    run_processor=AuthRequestProcessor(),
)

asyncio.run(agent.run("查询我的 ECS 实例列表，并在一台运行中的实例上执行 `uname -a`"))