构建 Digital FTE：4 小时速成课

15 个概念，覆盖 80% 的真实用法：Skills、System of Record 和 MCP

上一门课构建了 chat agent。本课把它提升为 Digital FTE：一个能携带能力、读写系统记录、通过明确边界访问业务数据、留下审计轨迹，并在必要时请求批准的 AI Worker。

更具体地说，上一课的 agent 仍然像一个会话应用：它有 session、有 tools、有 guardrails，但它的能力主要写在代码里，业务事实主要在 stub 或本地文件里。Digital FTE 的不同在于它开始拥有「员工」应有的基础：可携带的能力包、稳定的系统记录、可查询的知识库、明确的 runtime boundary、所有重要动作的 audit trail，以及超出权限时停下来请求批准的纪律。

Where this course sits in the Agent Factory thesis

Digital FTE 是 Agent Factory 的第一个真正 Worker。它不再只是对话，而是有职责、有记录、有工具边界、有持久 memory 的工作单元。

The 15-concept cheat sheet

组	概念	核心
Skills	1–5	能力作为可迁移文件夹。
Postgres + pgvector	6–10	系统记录、schema、embedding、audit。
MCP	11–15	agent 与系统记录之间的协议边界。

Who this course is for

适合已经能构建 basic agent，现在想让它处理真实业务工作的读者。你不必是数据库专家，但需要愿意让 agent 读写结构化数据。

What stays from the previous course, what's new

保留 Agent、Runner、tools、sessions、guardrails 和 tracing。新增 Skills、Neon Postgres、pgvector、embedding pipeline、MCP server、audit trail 和 approval discipline。

The fifteen-minute quick win: succeed once, then study why it worked

Quick Win 的目标是让 agent 通过 MCP 写一条 note，同时写一条 audit log。先成功一次，再回头理解为什么这个结构能扩展。

# Log a note
uv run python -m notes_mcp

# notes_mcp.py
import asyncpg
import json
import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("notes")
_pool = None

async def pool():
    global _pool
    if _pool is None:
        _pool = await asyncpg.create_pool(os.environ["DATABASE_URL"])
    return _pool

@mcp.tool()
async def save_note(text: str) -> dict[str, str]:
    """Save a note to the durable notes table. Returns the row_id."""
    p = await pool()
    async with p.acquire() as c:
        async with c.transaction():
            row_id = await c.fetchval(
                "INSERT INTO notes (note_text) VALUES ($1) RETURNING id::text", text)
            await c.execute(
                "INSERT INTO audit_log (action, payload) VALUES ($1, $2::jsonb)",
                "note_saved",
                json.dumps({"row_id": row_id, "len": len(text)}),
            )
    return {"row_id": row_id, "status": "saved"}

if __name__ == "__main__":
    mcp.run(transport="stdio")

# cli.py — Quick Win shape
from agents import Agent, Runner

agent = Agent(name="Notes worker", instructions="Use MCP tools to save notes.")
print(Runner.run_sync(agent, "Save a note saying the quick win worked.").final_output)

Quick Win 后要检查数据库，而不是只看 agent 回答：

select n.id, n.note_text, a.action, a.payload
from notes n
join audit_log a on (a.payload->>'row_id')::uuid = n.id
order by n.created_at desc
limit 1;

你学习的不是「MCP 可以调 tool」这么简单，而是一个 Worker 的最小生产形状：action 写业务表，同时 audit log 记录这次 action 发生过。

Part 1: Skills, capability as portable folders

Concept 1: What an Agent Skill is

Skill 是一个文件夹，通常包含 SKILL.md 和按需读取的参考文件或脚本。它把能力从 agent 代码中移出来，变成可发现、可版本化、可复用的包。

# Hello skill

Use this skill when the user asks for a short greeting.

更完整的 Skill 文件夹可以长这样：

summarize-ticket/
├── SKILL.md
├── references/
│   ├── tone-guide.md
│   └── escalation-policy.md
├── scripts/
│   └── normalize_ticket.py
└── assets/
    └── examples.json

SKILL.md 是入口。references、scripts 和 assets 只有在需要时才加载。这就是 Skill 能扩展而不把每一轮上下文撑爆的原因。

Concept 2: Progressive disclosure, the three-stage skill loading model

progressive disclosure 控制上下文成本。不是把所有知识一次性塞进 prompt，而是先暴露能力目录，真正需要时再加载细节。

三阶段模型：

1. Discovery。 启动时只加载所有 skills 的 name 和 description。成本随 skill 数量线性增长，但每个 skill 只占几十 token。
2. Activation。 当模型判断某个 skill 相关时，加载完整 SKILL.md。成本只发生在匹配 turn。
3. Execution。 SKILL.md 引用的 files/scripts/assets 按需读取。成本只发生在实际触达引用时。

假设有 30 个 skills，每个 description 80 tokens，一个被激活的 SKILL.md 1500 tokens，再读取一个 2000-token reference file，总成本约是 6900 tokens，而不是把 30 个 skills 的完整内容全塞进去的 165000 tokens。这就是 skill library 可负担的原因。

设计 consequence：description 是最关键字段。helps with PDFs 太模糊，会误触发；extracts text and tables from PDFs, fills PDF forms, merges PDFs; use when working with PDF files 才能在该触发时触发、不该触发时保持安静。

Concept 3: Writing a SKILL.md, the contract a skill makes with the model

SKILL.md 是能力契约：何时使用、输入是什么、输出是什么、不能做什么、需要读哪些文件、如何验证。写得越清楚，agent 越少猜。

一个好的 SKILL.md 应包含：

---
name: summarize-ticket
description: Use when turning raw support tickets into concise summaries with urgency, requested action, and escalation flags.
---

# Summarize Ticket

## When to use

Use this when the user gives a raw customer-support ticket or asks for the story of a ticket ID.

## Inputs

- raw customer message
- customer tier if available
- prior ticket context if available

## Output

Return:
- summary
- urgency
- requested action
- escalation flags

## Rules

- Preserve uncertain facts as uncertain.
- Do not invent order status.
- Escalate payment, legal, safety, or data-deletion requests.

## Verification

Check that every claim in the summary appears in the source ticket or retrieved context.

description 决定是否触发；body 决定触发后如何工作；references 承载深知识；scripts 承载确定性动作。

Concept 4: Skill packaging, where skills live and how they travel

Skills 可以放在项目、用户目录、组织库或 marketplace。关键是路径、版本、依赖和安全边界明确。

常见位置：

位置	适合什么
项目内 .claude/skills/ 或类似目录	和一个 repo 紧密绑定的能力
用户目录 skills	个人跨项目复用能力
组织 skills repo	团队共享能力
marketplace / curated registry	可分发、可安装、可版本化能力

Skill 能 travel，是因为它是文件，不是某个聊天会话中的隐形 prompt。真正的迁移单位是 SKILL.md 加 references/scripts/assets，而不是「我曾经在某个会话里告诉过 agent」。

Concept 5: Composing skills, when to chain small skills vs. write one big one

一个大 Skill 更容易发现，多个小 Skill 更容易复用和调试。高频、稳定、跨项目的能力适合小 Skill；一次性流程适合大 Skill。

单体 skill 的优点是发现简单、一次激活完成整个流程；缺点是上下文会在多步骤中累积，debug 较难。多个小 skills 的优点是可复用、每一步上下文更小、intermediate artifacts 可检查；缺点是激活次数更多，handoff 文件设计更重要。

对弱一些的模型（小模型、本地模型、便宜模型），还要额外写 orchestration scaffold。强模型通常能遵守 SKILL.md 中的「先 X 再 Y」；弱模型更容易丢步骤。经验法则：强模型的努力写进 SKILL.md，弱模型的努力写进 system prompt 的 short GENERAL-FLOW。

Part 2: Neon Postgres + pgvector as system of record

Concept 6: Why managed Postgres, and why Neon specifically

Worker 需要系统记录，而不是聊天记忆。Postgres 提供事务、schema、约束、索引、查询和审计。Neon 让学习路径更轻：serverless、branching、连接字符串简单。

Digital FTE 会被问到真实业务问题：「六周前我们怎么回复这个客户？」、「这个问题以前解决过吗？」、「这次退款是否重复？」这些问题不能靠模型记忆回答。Postgres 的价值是让事实结构化、可查询、可约束、可审计。Neon 不是唯一选择；它只是本课 worked example 中学习成本最低的 managed Postgres。

Concept 7: The Worker's schema, what tables an agent actually needs

最小 schema 包括 conversations、messages、documents、embeddings、audit_log 和 capability_invocations。它们把对话、知识、工具调用和审计连成一个系统记录。

最小 schema 形状：

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE conversations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id UUID,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE messages (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  conversation_id UUID REFERENCES conversations(id),
  role TEXT NOT NULL,
  body TEXT NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE documents (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  source TEXT NOT NULL,
  title TEXT NOT NULL,
  body TEXT NOT NULL,
  metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE embeddings (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  document_id UUID REFERENCES documents(id),
  chunk_text TEXT NOT NULL,
  embedding VECTOR(1536) NOT NULL,
  metadata JSONB NOT NULL DEFAULT '{}'::jsonb
);

CREATE TABLE audit_log (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  action TEXT NOT NULL,
  actor TEXT,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE capability_invocations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  capability_name TEXT NOT NULL,
  input JSONB NOT NULL,
  output JSONB,
  audit_log_id UUID REFERENCES audit_log(id),
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

customer-support worked example 会在这些基础表上增加 customers、orders、tickets、refunds 等领域表。

Concept 8: pgvector basics, types, distance operators, indexes

pgvector 让 Postgres 存储向量并做相似度搜索。你需要理解 VECTOR(1536)、cosine distance、HNSW index 和 top-k 查询。不要把它神秘化；它就是数据库里的相似度索引。

查询形状：

SELECT d.title, e.chunk_text
FROM embeddings e
JOIN documents d ON d.id = e.document_id
WHERE e.document_id IS NOT NULL
ORDER BY e.embedding <=> $1
LIMIT 5;

$1 是用户问题在查询时生成的 embedding。<=> 是 cosine distance operator；距离越小越相似。

索引选择：

CREATE INDEX idx_embeddings_hnsw
  ON embeddings USING hnsw (embedding vector_cosine_ops);

新项目默认从 HNSW 开始：build 慢一些，query 快，operator 支持广。IVFFlat build 更快但 query 慢，适合频繁重建索引的场景。很大规模才考虑 DiskANN/pgvectorscale。

Concept 9: The embedding pipeline, text in, queryable vector out

pipeline 包括 chunk、embed、store、index 和 query。常见错误是换了 embedding model、chunk 太大或太小、不同来源混在一起却没有 filter。

# src/chat_agent/embedding/chunker.py
def chunk_text(text: str, max_tokens: int = 400) -> list[str]:
    return [text]

# src/chat_agent/embedding/embedder.py
def embed(texts: list[str]) -> list[list[float]]:
    return [[0.0] * 1536 for _ in texts]

pipeline 的 discipline：

# src/chat_agent/embedding/seeder.py
def seed_document_library(rows: list[dict]) -> None:
    for row in rows:
        doc_id = insert_document(
            source="past_case",
            title=row["summary"],
            body=row["summary"] + "\n\n" + row["resolution"],
            metadata={"customer_email": row["customer_email"]},
        )
        chunks = chunk_text(row["summary"] + "\n\n" + row["resolution"])
        vectors = embed(chunks)
        insert_embeddings(doc_id, chunks, vectors)

三类 trap：insert-time 和 query-time 使用不同 embedding model；chunk size 与用户问题粒度不匹配；不同来源混合但查询时没有 metadata filter。

Concept 10: Audit trail as discipline, what "reads and writes" means for a Worker

audit trail 不是日志装饰。每次 tool invocation、数据库写入、approval request、外部 API 调用都应该写 audit row。生产里最重要的问题通常是：谁在什么时候基于什么输入做了什么？

# src/chat_agent/audit.py
def record_audit(actor: str, action: str, entity_id: str) -> None:
    pass

每个 meaningful action 应写两类事实：业务事实写入业务表，动作事实写入 audit_log。如果 tool 是 capability，还应写 capability_invocations。

INSERT INTO audit_log (action, actor, payload)
VALUES (
  'refund_issued',
  'customer_support_worker',
  jsonb_build_object(
    'order_id', $1,
    'amount_cents', $2,
    'reason', $3
  )
);

没有 audit row，未来你无法回答「agent 做了什么、何时做、基于什么输入做」。

Part 3: MCP, wiring the agent to the system of record

Concept 11: What MCP is and isn't

MCP 是 agent 和工具/数据之间的协议边界。它不是数据库，不是权限系统，也不是 magic memory。它把「可调用能力」暴露给 agent，同时把真实数据访问留在 server 端。

MCP 买到三件事：

• Scope： agent 只能调用 server 暴露的窄工具。
• Isolation： 数据库凭据、连接池、事务留在 server 内。
• Reusability： 同一个 MCP server 可被多个 agents、多个 runtimes 使用。

MCP 不替你做 authorization、audit、schema design 或 slow-query tuning。那些仍然是你的 server 责任。

Concept 12: The Neon MCP server, development plane, not runtime

Neon MCP server 适合开发时浏览 schema、运行迁移、理解数据库。生产 runtime 不应该让 agent 直接拿到宽权限数据库工具。

# In your project, run:
uvx neon-mcp-server

Claude Code 和 OpenCode 配置形状不同，但原则相同：

{
  "mcpServers": {
    "neon": {
      "command": "npx",
      "args": ["-y", "mcp-remote@latest", "https://mcp.neon.tech/mcp"]
    }
  }
}

Neon MCP 很强，因为它能做 Neon API 能做的很多事；这也正是它不适合 production runtime 的原因。生产 Worker 不应该拿到 general-purpose run_sql。

Concept 13: Connecting MCP to the OpenAI Agents SDK

SDK 可以通过 MCP server 暴露 tools。stdio 适合本地和子进程；streamable HTTP 适合服务化 runtime。

# tools/scratch_with_neon.py — development utility, not production

runtime 连接时要区分 stdio 和 streamable HTTP。stdio 适合本地子进程：

from agents.mcp import MCPServerStdio

server = MCPServerStdio(
    params={
        "command": "python",
        "args": ["-m", "customer_data_mcp.server"],
    }
)

HTTP 适合独立服务：

from agents.mcp import MCPServerStreamableHttp

server = MCPServerStreamableHttp(
    url="https://customer-data.example.com/mcp",
    headers={"Authorization": "Bearer ..."},
)

Concept 14: Custom MCP servers, when to write your own vs. when not to

如果 vendor 已有窄 scope MCP server，优先使用。如果你需要业务规则、审计、权限、事务和领域语义，就写 custom MCP server。

# servers/customer_data_mcp/server.py
# Run with: python -m customer_data_mcp.server
import asyncpg
from mcp.server.fastmcp import FastMCP
from pydantic import Field
from typing import Annotated

mcp = FastMCP("customer-data")

@mcp.tool()
async def get_customer(email: Annotated[str, Field(description="Customer email.")]) -> dict:
    """Return the customer profile for a support workflow."""
    ...

@mcp.tool()
async def find_similar_cases(query: str, limit: int = 5) -> list[dict]:
    """Search resolved cases by semantic similarity."""
    ...

@mcp.tool()
async def issue_refund(order_id: str, amount_cents: int, reason: str) -> dict:
    """Issue an authorized refund and write audit_log."""
    ...

decision tree 的优先级是：单个低风险本地函数用 @function_tool；vendor 已有窄 server 用 vendor MCP；需要多 agent 复用、长期连接、敏感 scope 或 process isolation 时，写 custom MCP server。

Concept 15: MCP under load: transports, pooling, and what happens at scale

生产 MCP server 要考虑连接池、超时、并发、backpressure、tool schema 稳定性和 audit。stdio 便于本地，HTTP 更适合多实例部署。

负载下要检查五层：

1. agent-side MCP client 是否重复创建连接；
2. transport 是否适合部署形态；
3. MCP server 内部 Postgres pool 是否太小；
4. Postgres 查询是否慢；
5. agent 与 MCP server 间是否有 network/DNS 问题。

症状对照：tools list 很慢，先检查 client caching；tool 调用排队，检查 pool；单个查询慢，跑 EXPLAIN ANALYZE；偶发 timeout，检查 network 和 retry policy。

Part 4: The worked example, customer-support Worker

The brief

构建一个客户支持 Worker：能总结 ticket、检索相似案例、访问客户系统记录、写 audit log、在越权时升级。

前置 setup：你有上一课的 starter chat-agent 项目（cli.py、agents.py、tools.py、models.py、guardrails.py、sandboxed.py）；你有 Neon account 并已认证；你已安装并登录 Claude Code 或 OpenCode。Part 4 不替换上一课项目，而是在其上叠加 Skills、system of record 和 MCP。

Decision 1: Update the rules file with the new architecture

# chat-agent

## Stack

Python、OpenAI Agents SDK、Neon Postgres、pgvector、MCP。

## Architecture (NEW)

Skills carry capabilities. Postgres is the system of record. MCP is the runtime boundary.

## Critical rules (NEW)

所有数据库写入都要 audit。生产 runtime 不使用宽权限 Neon MCP。

## Commands

uv run pytest

Decision 2: Plan the schema and the Skill set

先计划 tables、indexes、Skills 和 MCP tools。不要让 agent 一边猜 schema 一边写 runtime。

应产出一份 plan，包括：

Tables:
- customers
- orders
- tickets
- refunds
- documents
- embeddings
- audit_log
- capability_invocations

Skills:
- summarize-ticket
- find-similar-cases
- escalate-with-context

MCP tools:
- get_customer
- find_similar_cases
- issue_refund
- write_audit
- escalate_case

Decision 3: Provision Neon and run the schema migration

创建 Neon project、database、migration 和 pooled connection。确认 pgvector extension 可用。

领域表示例：

CREATE TABLE customers (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email TEXT NOT NULL UNIQUE,
    tier TEXT NOT NULL CHECK (tier IN ('free', 'pro', 'enterprise')),
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE orders (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    customer_id UUID NOT NULL REFERENCES customers(id) ON DELETE CASCADE,
    public_id TEXT NOT NULL UNIQUE,
    placed_at TIMESTAMPTZ NOT NULL,
    amount_cents INT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('placed','shipped','delivered','refunded','cancelled'))
);

CREATE TABLE tickets (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    customer_id UUID NOT NULL REFERENCES customers(id),
    summary TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('open','in_review','resolved','blocked')),
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Decision 4: Write the first Skill, summarize-ticket

# Summarize ticket

Use this skill to turn a raw support ticket into a concise issue summary.

## Example: short ticket

Input: one short customer message.
Output: summary, urgency, requested action.

## Example: complex ticket

Preserve timeline and uncertain facts.

## Edge cases

Escalate when legal, payment, or safety issues appear.

Decision 5: Build the embedding pipeline and seed the document library

把 resolved tickets 或政策文档分块、embedding、写入 documents 和 embeddings。用小数据集先验证 top-k 结果。

# data/seed/resolved-tickets.csv (columns: id, customer_email, summary, resolution)
id,customer_email,summary,resolution
T-1,maya@example.com,Duplicate charge,Issued refund after verification
T-2,sara@example.com,Refund not showing up two weeks after return approved,Refund was stuck in pending due to a payment-gateway batch delay; manually re-triggered and funds posted within 24h.
T-3,raj@example.com,Cannot log in after email change,Account email was updated but the session cache held the old address; cleared the session and the customer logged in normally.

seeder 要求：

1. Read the CSV.
2. For each row:
   a. Find or create customer by email.
   b. Insert ticket with status='resolved'.
   c. Insert document source='past_case'.
   d. Chunk document body.
   e. Embed chunks.
   f. Insert embeddings linked to documents.id.
   g. Write audit_log.

Decision 6: Write the custom MCP server for runtime access

custom MCP server 暴露窄工具：get_customer、find_similar_cases、write_audit、escalate_case。server 负责 SQL、权限和事务。

# src/customer_data_mcp/server.py
# The agent spawns the stdio MCP server itself (via MCPServerStdio), so you do
# not start it separately.

refund tool 必须在一个 transaction 中写业务事实和 audit：

@mcp.tool()
async def issue_refund(order_id: str, amount_cents: int, reason: str) -> dict[str, str]:
    pool = await get_pool()
    async with pool.acquire() as conn:
        async with conn.transaction():
            refund_id = await conn.fetchval(
                """INSERT INTO refunds (order_id, amount_cents, reason)
                   VALUES ($1::uuid, $2, $3) RETURNING id::text""",
                order_id, amount_cents, reason,
            )
            await conn.execute(
                "UPDATE orders SET status='refunded' WHERE id=$1::uuid",
                order_id,
            )
            await conn.execute(
                "INSERT INTO audit_log (action, payload) VALUES ($1, $2::jsonb)",
                "refund_issued",
                json.dumps({"refund_id": refund_id, "order_id": order_id}),
            )
    return {"refund_id": refund_id, "status": "issued"}

Decision 7: Wire audit logging everywhere

每次读取、检索、写入、升级和 approval request 都写 audit row。audit 不应依赖最终回答成功。

Decision 8: Verify end-to-end with the test scenario

给一个真实感 ticket，要求 Worker 总结、检索相似案例、提出回复草稿、写 audit，并在越权事项上升级。

验证 checklist：

□ Ticket summary uses summarize-ticket Skill.
□ Similar cases are retrieved from embeddings.
□ Customer data comes through custom MCP server, not broad Neon MCP.
□ Any refund action writes refunds + audit_log in one transaction.
□ Out-of-envelope request escalates instead of executing.
□ capability_invocations records every tool/skill call.

What just happened

你把 chat agent 变成了 Worker。它现在有 portable Skills、system of record、knowledge retrieval、runtime boundary 和 audit discipline。

Part 5: Where this course leaves off

Cost shape of a Worker: what does it actually cost to operate?

成本来自模型调用、embedding、数据库、MCP server、观测和人工审批。embedding 通常是批量成本，模型调用是持续成本，错误检索和无界上下文会放大费用。

典型成本结构：embedding 是 seed/update 时的批量成本；每次 customer interaction 的主要成本是模型 tokens；MCP server 和 Postgres 通常较小；人工 approval 是时间成本。真正会爆的是错误架构：每轮都塞完整知识库、重复 embedding、慢 query 导致 retries、宽权限 tool 造成多余调用。

Swap guide: the architecture is invariant, the products are not

Neon 可换成 Supabase、RDS 或 Azure Database for PostgreSQL；OpenAI embedding 可换成其他 embedding；MCP transport 可换。不可变的是 system-of-record、capability boundary 和 audit。

Product in course	可替换为	不变量
Neon Postgres	Supabase、RDS、Azure Database for PostgreSQL	relational system of record
pgvector	pgvectorscale、external vector DB	queryable semantic memory
OpenAI embeddings	其他 embedding model	same model at insert and query time
MCP stdio	streamable HTTP	narrow tool boundary
Agent Skills	兼容 skill folder system	portable capability

What this course doesn't cover (yet)

本课不处理 durable scheduling、large-scale deployment、workforce management、hiring API 或 owner delegation。后续课程逐步补齐。

How to actually get good at this

先从一个小 Worker 开始：一个 Skill、三张表、一个 MCP server、一个 audit log。跑通后再加 embedding、approval 和更多 Skills。不要一次构建完整公司。

The shift in what you spend time on

你花的时间会从「写 prompt」转向「定义 schema、边界、audit、eval 和 workflows」。这就是从聊天机器人到 Worker 的转变。

Quick reference

The 15 concepts in one line each

Skill 是 portable capability；progressive disclosure 控成本；SKILL.md 是契约；Skills 需要 packaging；composition 要权衡；managed Postgres 是系统记录；Worker 需要 schema；pgvector 做相似度；embedding pipeline 让文本可查询；audit 是纪律；MCP 是协议边界；Neon MCP 属于开发平面；SDK 可连接 MCP；custom MCP 用于业务 runtime；MCP under load 需要 pooling 和边界。

Decision tree: @function_tool vs. custom MCP server vs. vendor MCP server

单个、低风险、本地函数用 @function_tool。vendor 已有窄权限 server，用 vendor MCP。需要业务事务、audit 和权限时，写 custom MCP server。

File location quick-ref

Skills 放在 .claude/skills 或项目 skills 目录；embedding code 放在 src/.../embedding；MCP server 放在 src/..._mcp；schema migration 放在 migrations/；seed data 放在 data/seed/。

When something feels wrong

检索结果奇怪，检查 chunk 和 embedding model。agent 越权，检查 MCP tool scope。审计缺失，检查每个 tool 是否写 audit。数据库慢，检查 indexes 和 connection pooling。

常见症状：

• 「Skill 不触发」→ description 太窄或触发词不贴近用户真实说法。
• 「Skill 乱触发」→ description 太泛。
• 「pgvector 结果不相关」→ chunking 错、embedding model 不一致、metadata filter 缺失。
• 「MCP under load 慢」→ pool 太小、query 慢、client 没缓存 tools list。
• 「Neon MCP 在 production 里吓人」→ 因为它是 development plane；写 custom MCP server。

Appendix: Audience, prerequisites, glossary

A.1: Audience & prerequisites

适合已经完成 basic agent 课程的读者。需要 Python、SQL 基础、环境变量和命令行能力。

A.2: What the previous course taught you that this course assumes

你已经知道 Agent、Runner、tools、sessions、guardrails、tracing 和 human approval 的基本概念。

A.3: Postgres essentials this course uses

表、主键、外键、索引、事务、连接池、migration、extension 和 SQL 查询。

A.4: How to read this page on first pass

第一次先读概念和 Part 4 brief。第二次再跟着 Decisions 动手。不要在第一次阅读时卡在每个 SQL 细节上。

A.5: Glossary

Digital FTE、AI Worker、Skill、system of record、pgvector、embedding、MCP、audit trail、approval envelope。

A.6: What this appendix does NOT replace

它不替代 Postgres、pgvector、MCP 或 OpenAI Agents SDK 官方文档。它只是本课程的最小背景。