基本思路

你的 LangChain Agent 对任务进行推理。当它需要外部数据时,调用 ATHTool——通过 ATH 以适当的用户授权进行 API 调用。

ATHTool 实现

"""ATHTool — 将 ATH 代理转发的 API 封装为 LangChain 工具。"""

import json
from typing import Optional, Type
from pydantic import BaseModel, Field
from langchain_core.tools import BaseTool
from ath import ATHGatewayClient, ATHError


class ATHToolInput(BaseModel):
    method: str = Field(description="HTTP method: GET, POST, PUT, DELETE")
    path: str = Field(description="API path, e.g. /user/repos")
    body: Optional[str] = Field(default=None, description="JSON body for POST/PUT")


class ATHTool(BaseTool):
    """Call an API through ATH gateway with user-consented access."""

    name: str = "ath_api"
    description: str = "Call an external API securely through ATH."
    args_schema: Type[BaseModel] = ATHToolInput
    client: ATHGatewayClient
    provider: str

    class Config:
        arbitrary_types_allowed = True

    def _run(self, method: str, path: str, body: Optional[str] = None, **kwargs) -> str:
        try:
            parsed = json.loads(body) if body else None
            result = self.client.proxy(self.provider, method.upper(), path, body=parsed)
            return json.dumps(result, indent=2) if isinstance(result, dict) else str(result)
        except ATHError as e:
            return json.dumps({"error": e.code, "message": e.message})

配合演示使用

from ath import ATHGatewayClient
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

# 设置 ATH 客户端(假设注册 + 令牌已完成)
client = ATHGatewayClient(url="https://your-gateway.com", agent_id=..., private_key=pem)
client.load_credentials("creds.json")
client.set_token("ath_tk_...")

# 创建工具
shop_tool = ATHTool(
    name="shop_api",
    description="Access ATH Shop: GET /products, POST /cart/add, POST /orders, GET /orders",
    client=client,
    provider="ath-shop",
)

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
    ("system", "You help users shop. Use shop_api to browse and buy products."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, [shop_tool], prompt)
executor = AgentExecutor(agent=agent, tools=[shop_tool])

# 运行
result = executor.invoke({"input": "Find me a laptop under $1000 and add it to my cart"})
print(result["output"])

安全保障

LangChain Agent 只能访问注册时批准用户授权的内容。如果它尝试访问没有权限的作用域:
# Agent 尝试访问订单但只有 products:read 权限
result = client.proxy("ath-shop", "GET", "/orders")
# → ATHError(code="SCOPE_NOT_APPROVED")
Agent 无法自行提升权限。