以编程方式使用文档

以编程方式使用 Fleet 代理主要有两种方法：

  • - **从代码调用**：通过 LangGraph SDK 或 REST API 远程调用您的代理，无需下载任何内容。
  • - **导出到代码**：下载您的代理配置，并使用 fleet-deepagents-export package.

从代码调用

您可以使用 LangGraph SDK 或 REST API 从您的应用程序调用 LangSmith Fleet 代理。Fleet 代理在 Agent Server上运行，因此您可以使用与其他 LangSmith 部署.

REST API 允许您从任何支持 HTTP 请求的语言或平台调用您的代理。

前提条件

    pip install langgraph-sdk python-dotenv
    
    yarn add @langchain/langgraph-sdk

### 身份验证 要向您代理的 Fleet 部署进行身份验证，请提供 LangSmith 个人访问令牌 (PAT) 给 api_key 参数，在实例化 LangGraph SDK 客户端时，或通过 X-API-Key 标头。如果使用 X-API-Key您还必须设置 X-Auth-Scheme 标头为 langsmith-api-key.

如果您传递的 PAT 与代理的所有者无关，您的请求将被拒绝，并显示 404 Not Found error.

如果您尝试调用的代理是 工作区代理 且您不是所有者，则可以执行与在 UI 中相同的所有操作（只读）。

1. 获取代理 ID 和 URL

要获取您代理的 agent_id 和 api_url:

  1. 在 LangSmith UI中，导航到您代理的收件箱。
  2. 在代理名称旁边，点击 **编辑代理** icon.
  3. 点击 **设置** 右上角的图标。
  4. 点击 **查看代码片段** 查看您代理的预填充值。

复制以下代码并替换 agent_id 和 api_url 为代理代码片段中的值。

在项目根目录中创建一个 .env 文件，包含您的 个人访问令牌:

LANGGRAPH_API_KEY=your-personal-access-token

2. 获取智能体配置

通过获取您的智能体配置来验证连接：

Python

from dotenv import load_dotenv
from langgraph_sdk.client import get_client

load_dotenv()

agent_id = "your-agent-id"

api_key = os.getenv("LANGGRAPH_API_KEY")
api_url = ".us.langgraph.app"

client = get_client(
    url=api_url,
    api_key=api_key,
    headers={
        "X-Auth-Scheme": "langsmith-api-key",
    },
)

async def get_assistant(agent_id: str):
    agent = await client.assistants.get(agent_id)
    print(agent)

if __name__ == "__main__":

    asyncio.run(get_assistant(agent_id))

TypeScript

const agentId = "your-agent-id";

const apiKey = process.env.LANGGRAPH_API_KEY;
const apiUrl = ".us.langgraph.app";

const client = new Client({
  apiUrl,
  apiKey,
  defaultHeaders: {
    "X-Auth-Scheme": "langsmith-api-key",
  },
});

async function main(agentId: string) {
  const agent = await client.assistants.get(agentId);
  console.log(agent);
}

main(agentId).catch(console.error);

cURL

curl --request GET \
    --url ".us.langgraph.app/assistants/your-agent-id" \
    --header 'Content-Type: application/json' \
    --header 'X-Api-Key: your-personal-access-token' \
    --header 'X-Auth-Scheme: langsmith-api-key'

3. 调用智能体

以下示例展示了如何向您的智能体发送消息并接收响应。您可以使用 **无状态** 运行（无线程，无对话历史）或 **有状态** 运行（使用线程在多次交互中保持对话历史）。

无状态运行

无状态运行发送单个请求并返回完整响应。不保留对话历史。这是调用智能体的最简单方式：

Python

from dotenv import load_dotenv
from langgraph_sdk.client import get_client

load_dotenv()

agent_id = "your-agent-id"

api_key = os.getenv("LANGGRAPH_API_KEY")
api_url = "https://.us.langgraph.app"

client = get_client(
    url=api_url,
    api_key=api_key,
    headers={
        "X-Auth-Scheme": "langsmith-api-key",
    },
)

result = await client.runs.wait(
    None,
    agent_id,
    input={
        "messages": [
            {"role": "user", "content": "What can you help me with?"}
        ]
    },
)
print(result)

TypeScript

const agentId = "your-agent-id";

const apiKey = process.env.LANGGRAPH_API_KEY;
const apiUrl = ".us.langgraph.app";

const client = new Client({
  apiUrl,
  apiKey,
  defaultHeaders: {
    "X-Auth-Scheme": "langsmith-api-key",
  },
});

const result = await client.runs.wait(
  null,
  agentId,
  {
    input: {
      messages: [
        { role: "user", content: "What can you help me with?" }
      ]
    }
  }
);
console.log(result);

cURL

curl --request POST \
    --url ".us.langgraph.app/runs/wait" \
    --header 'Content-Type: application/json' \
    --header 'X-Api-Key: your-personal-access-token' \
    --header 'X-Auth-Scheme: langsmith-api-key' \
    --data '{
        "assistant_id": "your-agent-id",
        "input": {
            "messages": [
                {
                    "role": "user",
                    "content": "What can you help me with?"
                }
            ]
        }
    }'

无状态流式运行

要在生成响应时进行流式传输而不是等待完整结果，请使用流式端点：

Python

async for chunk in client.runs.stream(
    None,
    agent_id,
    input={
        "messages": [
            {"role": "user", "content": "What can you help me with?"}
        ]
    },
    stream_mode="updates",
):
    if chunk.data and "run_id" not in chunk.data:
        print(chunk.data)

TypeScript

const streamResponse = client.runs.stream(
  null,
  agentId,
  {
    input: {
      messages: [
        { role: "user", content: "What can you help me with?" }
      ]
    },
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  if (chunk.data && !("run_id" in chunk.data)) {
    console.log(chunk.data);
  }
}

cURL

curl --request POST \
    --url ".us.langgraph.app/runs/stream" \
    --header 'Content-Type: application/json' \
    --header 'X-Api-Key: your-personal-access-token' \
    --header 'X-Auth-Scheme: langsmith-api-key' \
    --data '{
        "assistant_id": "your-agent-id",
        "input": {
            "messages": [
                {
                    "role": "user",
                    "content": "What can you help me with?"
                }
            ]
        },
        "stream_mode": [
            "updates"
        ]
    }'

使用线程的有状态运行

要在多次交互中保持对话历史，请先创建一个线程，然后在其上运行您的智能体。同一线程上的每次后续运行都可以访问完整的消息历史：

Python

from dotenv import load_dotenv
from langgraph_sdk.client import get_client

load_dotenv()

agent_id = "your-agent-id"

api_key = os.getenv("LANGGRAPH_API_KEY")
api_url = ".us.langgraph.app"

client = get_client(
    url=api_url,
    api_key=api_key,
    headers={
        "X-Auth-Scheme": "langsmith-api-key",
    },
)

thread = await client.threads.create()

async for chunk in client.runs.stream(
    thread["thread_id"],
    agent_id,
    input={
        "messages": [
            {"role": "user", "content": "Hi, my name is Alice."}
        ]
    },
    stream_mode="updates",
):
    if chunk.data and "run_id" not in chunk.data:
        print(chunk.data)

async for chunk in client.runs.stream(
    thread["thread_id"],
    agent_id,
    input={
        "messages": [
            {"role": "user", "content": "What is my name?"}
        ]
    },
    stream_mode="updates",
):
    if chunk.data and "run_id" not in chunk.data:
        print(chunk.data)

TypeScript

const agentId = "your-agent-id";

const apiKey = process.env.LANGGRAPH_API_KEY;
const apiUrl = ".us.langgraph.app";

const client = new Client({
  apiUrl,
  apiKey,
  defaultHeaders: {
    "X-Auth-Scheme": "langsmith-api-key",
  },
});

const thread = await client.threads.create();

let streamResponse = client.runs.stream(
  thread["thread_id"],
  agentId,
  {
    input: {
      messages: [
        { role: "user", content: "Hi, my name is Alice." }
      ]
    },
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  if (chunk.data && !("run_id" in chunk.data)) {
    console.log(chunk.data);
  }
}

streamResponse = client.runs.stream(
  thread["thread_id"],
  agentId,
  {
    input: {
      messages: [
        { role: "user", content: "What is my name?" }
      ]
    },
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  if (chunk.data && !("run_id" in chunk.data)) {
    console.log(chunk.data);
  }
}

cURL

首先，创建一个线程：

curl --request POST \
    --url ".us.langgraph.app/threads" \
    --header 'Content-Type: application/json' \
    --header 'X-Api-Key: your-personal-access-token' \
    --header 'X-Auth-Scheme: langsmith-api-key' \
    --data '{}'

使用响应中的 thread_id 在线程上发送消息：

curl --request POST \
    --url ".us.langgraph.app/threads//runs/stream" \
    --header 'Content-Type: application/json' \
    --header 'X-Api-Key: your-personal-access-token' \
    --header 'X-Auth-Scheme: langsmith-api-key' \
    --data '{
        "assistant_id": "your-agent-id",
        "input": {
            "messages": [
                {
                    "role": "user",
                    "content": "Hi, my name is Alice."
                }
            ]
        },
        "stream_mode": [
            "updates"
        ]
    }'

在同一个线程上发送后续消息：

curl --request POST \
    --url ".us.langgraph.app/threads//runs/stream" \
    --header 'Content-Type: application/json' \
    --header 'X-Api-Key: your-personal-access-token' \
    --header 'X-Auth-Scheme: langsmith-api-key' \
    --data '{
        "assistant_id": "your-agent-id",
        "input": {
            "messages": [
                {
                    "role": "user",
                    "content": "What is my name?"
                }
            ]
        },
        "stream_mode": [
            "updates"
        ]
    }'

REST API 参考

下表总结了关键端点。将 `` 替换为您的智能体部署 URL。

操作方法端点
获取智能体信息GET/assistants/
创建线程POST/threads
运行（等待结果）POST/runs/wait
运行（流式）POST/runs/stream
在线程上运行（等待）POST/threads//runs/wait
/langsmith/agent-server-api/thread-runs/create-run-stream-outputPOST/threads//runs/stream

所有端点都需要以下标头： - Content-Type: application/json - X-Api-Key: 您的 个人访问令牌 - X-Auth-Scheme: langsmith-api-key

完整的 API 规范，请参阅 智能体服务器 API 参考.

导出为代码

导出为代码 功能允许您将 Fleet 智能体下载为独立运行的 Python 项目并在本地执行。当您想要以下操作时，这非常有用：

  • - 在您自己的基础设施中运行智能体，而无需调用 Fleet API
  • - 在 Fleet UI 支持的范围之外扩展或自定义智能体（添加自定义工具、中间件或技能）
  • - 检查或版本控制完整的智能体实现
  • - 使用 LangGraph Studio 进行本地开发和图形检查

fleet-deepagents-export 包（GitHub) 负责读取导出的配置，并将您的代理与 MCP 工具、子代理和技能连接起来。

前提条件

  • - Python 3.11+
  • - uv （推荐）用于依赖管理
  • - 一个要导出的 LangSmith Fleet 代理

1. 复制入门项目

位于的入门项目 examples/template-agent/ 是推荐的起点。克隆仓库并复制入门项目：

git clone https://github.com/langchain-ai/fleet-deepagents-export.git
cp -R fleet-deepagents-export/examples/template-agent my-agent
cd my-agent

2. 从 Fleet 导出您的代理

在 LangSmith UI中，打开您的代理并将其导出为 .zip file.

!fleet-export-code

然后将内容放入您入门项目的 fleet/ 目录中：

unzip path/to/my-export.zip -d fleet/

该 fleet/ 目录包含您的代理所需的一切： - AGENTS.md — 系统提示词 - config.json — 模型配置和工作区元数据 - tools.json — MCP 服务器连接 - subagents/ （可选）— 子代理定义 - skills/ （可选）— 技能指令

3. 配置您的环境

复制示例环境文件并填写所需的值：

cp .env.example .env

这三个 LANGSMITH_*_ID 值位于 fleet/config.json 下的 metadata。打开该文件并复制 tenant_id, organization_id和 ls_user_id 到您的 .env:

# Model provider — set the key for whichever provider your agent uses
ANTHROPIC_API_KEY=your-anthropic-api-key

# LangSmith credentials — copy IDs from fleet/config.json → metadata
LANGSMITH_API_KEY=your-langsmith-pat
LANGSMITH_TENANT_ID=your-tenant-id
LANGSMITH_ORGANIZATION_ID=your-organization-id
LANGSMITH_USER_ID=your-user-id       # required if your agent uses OAuth tools

# Built-in MCP tools (Gmail, Calendar, GitHub)
BUILTIN_MCP_URL=https://tools.langchain.com/mcp

4. 安装依赖并运行

make setup    # installs dependencies via uv sync

然后选择如何与您的代理交互：

make dev    # LangGraph Studio — browser UI for chat and graph inspection
make run    # terminal REPL via cli.py — text-only chat

5. 自定义代理

入门项目将 Fleet 拥有的文件与您可以自由编辑的文件分开：

File / DirectoryOwnerPurpose
fleet/Fleet在此放置导出的内容。重新解压以更新；其他内容不受影响。
agent.py您图接线。通过替换 model = components.pop("model") 行来覆盖模型。
custom_tools.py您添加代码定义的工具；运行时与 Fleet MCP 工具合并。
custom_middleware.py您添加 AgentMiddleware instances for logging, filters, pre/post hooks, etc.
custom_skills/您放置 <skill-name>/SKILL.md 文件；覆盖在 fleet/skills/.
cli.py您终端 REPL；自由编辑。

以下是完整的 agent.py 来自入门项目：

"""Standalone deepagent exported from LangSmith Fleet.

LangGraph Studio / dev server:  make dev
Terminal:                        make run  (see cli.py)

Extension points (edit these, not this file):
- ``custom_tools.py``      — add code-defined tools
- ``custom_middleware.py`` — wrap the agent loop with logging, filters, etc.
- ``custom_skills/``       — drop ``<skill-name>/SKILL.md`` files
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

from dotenv import load_dotenv

load_dotenv()

from custom_middleware import custom_middleware
from custom_tools import custom_tools
from deepagents import create_deep_agent
from fleet_deepagents_export import StaticSkillsLoader, load_agent_components

PROJECT_DIR = Path(__file__).parent
FLEET_DIR = PROJECT_DIR / "fleet"
CUSTOM_SKILLS_DIR = PROJECT_DIR / "custom_skills"

# Read SKILL.md from disk once; middleware injects into state on first turn.
_SKILL_LOADER = StaticSkillsLoader(
    [
        (FLEET_DIR / "skills", "/skills/fleet"),
        (CUSTOM_SKILLS_DIR, "/skills/custom"),
    ]
)


async def graph(runtime: Any):
    """Build and return the agent graph."""
    components = await load_agent_components(FLEET_DIR)
    model = components.pop("model")  # from fleet/config.json; replace to override
    components["tools"] = list(components["tools"]) + list(custom_tools)

    if _SKILL_LOADER.files:
        components["skills"] = _SKILL_LOADER.skill_paths

    return create_deep_agent(
        model=model,
        middleware=[_SKILL_LOADER, *custom_middleware],
        **components,
    ).with_config({"recursion_limit": 1000})

Re-exporting

当您从 Fleet 导出新版本的代理时，只需清除并重新解压 — 您的自定义内容不会受影响：

rm -rf fleet && unzip path/to/my-new-export.zip -d fleet/

支持的模型提供商

入门项目自带 langchain-anthropic, langchain-openai和 langchain-google-genai。对于任何其他提供商（例如 bedrock, fireworks），添加相应的 langchain-<provider> 包到 pyproject.toml.

MCP 身份验证

启动时，每个工具的 mcp_server_url 会根据 LangSmith 的 MCP 服务器注册表进行解析：

  • 内置 LangSmith 工具 （Gmail、日历、GitHub）—— 通过您的 LANGSMITH_API_KEY.
  • 静态凭证服务器 (auth_type: "headers"）—— 凭证来自注册表记录。需要 mcp-servers:invoke permission.
  • OAuth 服务器 (auth_type: "oauth"）—— 承载令牌从 LangSmith 的 OAuth 代理获取。首次运行时，系统会为任何尚未授权的每用户服务器打开浏览器窗口。