跳转到内容

代理式AI地牢游戏

模块 2:游戏 API 与库存 MCP 服务实现

Section titled “模块 2:游戏 API 与库存 MCP 服务实现”

我们将从实现游戏 API 开始。为此需要创建 5 个 API 接口:

  1. saveGame - 创建或更新游戏
  2. queryGames - 返回分页的历史游戏列表
  3. saveAction - 保存指定游戏的操作记录
  4. queryActions - 返回分页的指定游戏所有操作记录
  5. queryInventory - 返回分页的玩家库存物品列表

使用 Zodpackages/game-api/src/schema 目录下定义 API 输入输出模式:

import { z } from 'zod';
export const ActionSchema = z.object({
playerName: z.string(),
timestamp: z.iso.datetime(),
role: z.enum(['assistant', 'user']),
content: z.string(),
});
export type IAction = z.TypeOf<typeof ActionSchema>;

可以删除 packages/game-api/src/schema/echo.ts 文件,因为本项目不会使用它。

应用的实体关系图如下:

地牢冒险ER图

我们将使用 DynamoDB 实现数据库,并通过 ElectroDB 客户端库简化操作。首先需要运行以下命令安装 electrodb 和 DynamoDB 客户端:

Terminal window
pnpm add -w electrodb@3.5.0 @aws-sdk/client-dynamodb@3.913.0

现在根据上述 ER 图,在 packages/game-api/src/entities 目录下创建以下文件定义 ElectroDB 实体:

import { Entity } from 'electrodb';
import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
export const createActionEntity = (client?: DynamoDBClient) =>
new Entity(
{
model: {
entity: 'Action',
version: '1',
service: 'game',
},
attributes: {
playerName: { type: 'string', required: true, readOnly: true },
timestamp: {
type: 'string',
required: true,
readOnly: true,
set: () => new Date().toISOString(),
default: () => new Date().toISOString(),
},
role: { type: 'string', required: true, readOnly: true },
content: { type: 'string', required: true, readOnly: true },
},
indexes: {
primary: {
pk: { field: 'pk', composite: ['playerName'] },
sk: { field: 'sk', composite: ['timestamp'] },
},
},
},
{ client, table: process.env.TABLE_NAME },
);

ElectroDB 不仅可以定义类型,还能为时间戳等字段提供默认值。此外,ElectroDB 遵循 DynamoDB 最佳实践的单表设计

为了让 MCP 服务能与库存交互,请确保在 packages/game-api/src/index.ts 中导出库存实体:

export type { AppRouter } from './router.js';
export { appRouter } from './router.js';
export type { Context } from './init.js';
export * from './client/index.js';
export * from './schema/index.js';
export * from './entities/inventory.js';

将 DynamoDB 客户端注入 tRPC 上下文

Section titled “将 DynamoDB 客户端注入 tRPC 上下文”

由于每个操作流程都需要访问 DynamoDB 客户端,我们需要创建单一实例并通过上下文传递。在 packages/game-api/src 目录下进行以下修改:

import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
import { initTRPC } from '@trpc/server';
export interface IDynamoDBContext {
dynamoDb?: DynamoDBClient;
}
export const createDynamoDBPlugin = () => {
const t = initTRPC.context<IDynamoDBContext>().create();
return t.procedure.use(async (opts) => {
const dynamoDb = new DynamoDBClient();
const response = await opts.next({
ctx: {
...opts.ctx,
dynamoDb,
},
});
return response;
});
};

这是一个用于创建 DynamoDBClient 并注入上下文的插件。

现在开始实现 API 方法。在 packages/game-api/src/procedures 目录下进行以下修改:

import { createActionEntity } from '../entities/action.js';
import {
ActionSchema,
IAction,
QueryInputSchema,
createPaginatedQueryOutput,
} from '../schema/index.js';
import { publicProcedure } from '../init.js';
import { z } from 'zod';
export const queryActions = publicProcedure
.input(QueryInputSchema.extend({ playerName: z.string() }))
.output(createPaginatedQueryOutput(ActionSchema))
.query(async ({ input, ctx }) => {
const actionEntity = createActionEntity(ctx.dynamoDb);
const result = await actionEntity.query
.primary({ playerName: input.playerName })
.go({ cursor: input.cursor, count: input.limit });
return {
items: result.data as IAction[],
cursor: result.cursor,
};
});
import { ActionSchema, IAction } from '../schema/index.js';
import { publicProcedure } from '../init.js';
import { createActionEntity } from '../entities/action.js';
import { createGameEntity } from '../entities/game.js';
export const saveAction = publicProcedure
.input(ActionSchema.omit({ timestamp: true }))
.output(ActionSchema)
.mutation(async ({ input, ctx }) => {
const actionEntity = createActionEntity(ctx.dynamoDb);
const gameEntity = createGameEntity(ctx.dynamoDb);
const action = await actionEntity.put(input).go();
await gameEntity
.update({ playerName: input.playerName })
.set({ lastUpdated: action.data.timestamp })
.go();
return action.data as IAction;
});

可以删除 packages/game-api/src/procedures 中的 echo.ts 文件,因为本项目不会使用。

完成操作流程定义后,现在将其接入 API。更新以下文件:

import {
awsLambdaRequestHandler,
CreateAWSLambdaContextOptions,
} from '@trpc/server/adapters/aws-lambda';
import { t } from './init.js';
import { APIGatewayProxyEvent } from 'aws-lambda';
import { queryActions } from './procedures/query-actions.js';
import { saveAction } from './procedures/save-action.js';
import { queryGames } from './procedures/query-games.js';
import { saveGame } from './procedures/save-game.js';
import { queryInventory } from './procedures/query-inventory.js';
export const router = t.router;
export const appRouter = router({
actions: router({
query: queryActions,
save: saveAction,
}),
games: router({
query: queryGames,
save: saveGame,
}),
inventory: router({
query: queryInventory,
}),
});
export const handler = awsLambdaRequestHandler({
router: appRouter,
createContext: (
ctx: CreateAWSLambdaContextOptions<APIGatewayProxyEvent>,
) => ctx,
responseMeta: () => ({
headers: {
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': '*',
},
}),
});
export type AppRouter = typeof appRouter;

现在创建 MCP 服务,使智能体能管理玩家库存物品。

我们将为智能体定义以下工具:

  • list-inventory-items 获取玩家当前库存物品
  • add-to-inventory 向库存添加物品
  • remove-from-inventory 从库存移除物品

为节省时间,我们将内联定义所有工具:

import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import z from 'zod-v3';
import { createInventoryEntity } from ':dungeon-adventure/game-api';
/**
* Create the MCP Server
*/
export const createServer = () => {
const server = new McpServer({
name: 'inventory-mcp-server',
version: '1.0.0',
});
const dynamoDb = new DynamoDBClient();
const inventory = createInventoryEntity(dynamoDb);
server.tool(
'list-inventory-items',
"List items in the player's inventory. Leave cursor blank unless you are requesting subsequent pages",
{
playerName: z.string(),
cursor: z.string().optional(),
},
async ({ playerName }) => {
const results = await inventory.query
.primary({
playerName,
})
.go();
return {
content: [{ type: 'text', text: JSON.stringify(results) }],
};
},
);
server.tool(
'add-to-inventory',
"Add an item to the player's inventory. Quantity defaults to 1 if omitted.",
{
playerName: z.string(),
itemName: z.string(),
emoji: z.string(),
quantity: z.number().optional().default(1),
},
async ({ playerName, itemName, emoji, quantity = 1 }) => {
await inventory
.put({
playerName,
itemName,
quantity,
emoji,
})
.go();
return {
content: [
{
type: 'text',
text: `Added ${itemName} (x${quantity}) to inventory`,
},
],
};
},
);
server.tool(
'remove-from-inventory',
"Remove an item from the player's inventory. If quantity is omitted, all items are removed.",
{
playerName: z.string(),
itemName: z.string(),
quantity: z.number().optional(),
},
async ({ playerName, itemName, quantity }) => {
// If quantity is omitted, remove the entire item
if (quantity === undefined) {
try {
await inventory.delete({ playerName, itemName }).go();
return {
content: [
{ type: 'text', text: `${itemName} removed from inventory.` },
],
} as const;
} catch {
return {
content: [
{ type: 'text', text: `${itemName} not found in inventory` },
],
} as const;
}
}
// If quantity is specified, fetch current quantity and update
const item = await inventory.get({ playerName, itemName }).go();
if (!item.data) {
return {
content: [
{ type: 'text', text: `${itemName} not found in inventory` },
],
} as const;
}
const newQuantity = item.data.quantity - quantity;
if (newQuantity <= 0) {
await inventory.delete({ playerName, itemName }).go();
return {
content: [
{ type: 'text', text: `${itemName} removed from inventory.` },
],
} as const;
}
await inventory
.put({
playerName,
itemName,
quantity: newQuantity,
emoji: item.data.emoji,
})
.go();
return {
content: [
{
type: 'text',
text: `Removed ${itemName} (x${quantity}) from inventory. ${newQuantity} remaining.`,
},
],
};
},
);
return server;
};

随着工具数量增加,可以按需将其重构到独立文件中。

现在可以删除 packages/inventory/src/mcp-server 中未使用的 toolsresources 目录。

最后更新基础设施以创建 DynamoDB 表并授予 Game API 操作权限。按以下方式更新 packages/infra/src

import { CfnOutput } from 'aws-cdk-lib';
import {
AttributeType,
BillingMode,
ProjectionType,
Table,
TableProps,
} from 'aws-cdk-lib/aws-dynamodb';
import { Construct } from 'constructs';
import { suppressRules } from ':dungeon-adventure/common-constructs';
export type ElectrodbDynamoTableProps = Omit<
TableProps,
'partitionKey' | 'sortKey' | 'billingMode'
>;
export class ElectrodbDynamoTable extends Table {
constructor(scope: Construct, id: string, props?: ElectrodbDynamoTableProps) {
super(scope, id, {
partitionKey: {
name: 'pk',
type: AttributeType.STRING,
},
sortKey: {
name: 'sk',
type: AttributeType.STRING,
},
billingMode: BillingMode.PAY_PER_REQUEST,
...props,
});
this.addGlobalSecondaryIndex({
indexName: 'gsi1pk-gsi1sk-index',
partitionKey: {
name: 'gsi1pk',
type: AttributeType.STRING,
},
sortKey: {
name: 'gsi1sk',
type: AttributeType.STRING,
},
projectionType: ProjectionType.ALL,
});
// Suppress checkov rules that expect a KMS customer managed key and backup to be enabled
suppressRules(this, ['CKV_AWS_119', 'CKV_AWS_28'], 'No need for custom encryption or backup');
new CfnOutput(this, 'TableName', { value: this.tableName });
}
}

首先构建代码库:

Terminal window
pnpm nx run-many --target build --all

现在可以通过以下命令部署应用:

Terminal window
pnpm nx deploy infra dungeon-adventure-infra-sandbox/*

首次部署约需 8 分钟,后续部署约需 2 分钟。

部署完成后会看到类似以下输出(部分值已脱敏):

Terminal window
dungeon-adventure-sandbox-Application
dungeon-adventure-sandbox-Application: deploying... [2/2]
dungeon-adventure-sandbox-Application
Deployment time: 354s
Outputs:
dungeon-adventure-sandbox-Application.ElectroDbTableTableNameXXX = dungeon-adventure-sandbox-Application-ElectroDbTableXXX-YYY
dungeon-adventure-sandbox-Application.GameApiEndpointXXX = https://xxx.execute-api.region.amazonaws.com/prod/
dungeon-adventure-sandbox-Application.GameUIDistributionDomainNameXXX = xxx.cloudfront.net
dungeon-adventure-sandbox-Application.StoryApiEndpointXXX = https://xxx.execute-api.region.amazonaws.com/prod/
dungeon-adventure-sandbox-Application.UserIdentityUserIdentityIdentityPoolIdXXX = region:xxx
dungeon-adventure-sandbox-Application.UserIdentityUserIdentityUserPoolIdXXX = region_xxx

可通过以下方式测试 API:

  • 启动 tRPC 后端本地实例并使用 curl 调用 API
  • 使用支持 sigv4 签名的 curl 调用已部署 API

运行以下命令启动本地 game-api 服务:

Terminal window
TABLE_NAME=dungeon-adventure-infra-sandbox-Application-ElectroDbTableXXX-YYY pnpm nx run @dungeon-adventure/game-api:serve

服务启动后可通过以下命令调用:

Terminal window
curl -X GET 'http://localhost:2022/games.query?input=%7B%7D'

若命令执行成功,将看到如下响应:

{"result":{"data":{"items":[],"cursor":null}}}

恭喜,您已成功使用 tRPC 构建并部署了第一个 API!🎉🎉🎉