同一份类型化业务模型,AI 代理与开发者都是一级消费者。
MCP 是原生协议:强类型,底层是 GraphQL。
只需编写 SQLModel 实体和类型化 DTO,这就是全部工作量。
大模型生成代码很快——但缺乏结构约束时,技术债会在几周后浮现:逻辑重复、组件互相渗透、调试靠猜。业界称之为 vibe coding 的代价。
nexusx 把 AI 的书写面收窄为声明式模型——实体、关系与类型化用例方法。结构不靠 AI 发挥,由模型保证。
AI 只写模型与用例方法,不写散落各处的胶水代码——diff 小,人可审。
业务规则改一处,所有协议同步更新——维护成本不随交付协议数翻倍。
类型化契约,加上 Voyager:实体、关系、用例及其依赖渲染为一张实时 ER 图——不用先读代码就能掌握整个项目,无论你是新加入的人,还是一个新的 AI 会话。 Voyager →
语义级同构 —— 每种协议都从同一个类型化模型生成,而不是在复制品外面套一层包装。
# "list sprints" — written once per protocol
@app.get("/sprints")
async def rest_list_sprints() -> list[SprintOut]:
... # query + assembly, again
@strawberry.field
async def graphql_sprints(self) -> list[SprintType]:
... # types + loaders, again
@mcp.tool()
async def sprints_for_agents() -> str:
... # JSON dumping, again
# ↑ change the rule → fix every copy
class SprintService(UseCaseService):
"""Sprint planning operations."""
@query
async def list_sprints(cls) -> list[SprintSummary]:
"""List sprints with tasks, owners, and task count."""
return await load_sprints()
# six deliveries, one model ↓
类型化 FastAPI 路由,进入 OpenAPI。
create_use_case_router(api)
实体成为 by_id / by_filter 查询根,浏览切片关联数据。
Sprint { by_filter(limit: 10) { ... } }
用例方法经 compose schema 成为类型化字段。
compose_query(app, query, args)
AI 代理渐进式发现。
create_use_case_graphql_mcp_server([api])
服务即命令组。
list_sprints --select "name task_count"
从 compose schema 生成类型化客户端。
sprintService.listSprints()
两个 GraphQL 面,各司其职 —— 任选其一,或两者并用。
SQLModel 实体与关系变成 by_id / by_filter 查询根。无需编写关系 resolver —— DataLoader 批量加载保证遍历全程无 N+1。
GraphQLHandler类型化业务方法向 Web 客户端、集成方与 AI 代理暴露稳定的业务能力 —— 一份定义,REST / MCP / CLI / SDK 多端服务。
UseCaseService实体不等于 API 契约。DefineSubset 隐藏内部列、自动加载关系、计算派生字段。
# Per-endpoint: manual SQL, N+1, dict munging
async def get_sprints():
sprints = await session.exec(select(Sprint))
result = []
for s in sprints:
tasks = await session.exec(
select(Task).where(Task.sprint_id == s.id))
for t in tasks:
t.owner = await session.get(User, t.owner_id)
# N+1 queries, fragile dict construction
from nexusx import DefineSubset, ErManager, build_dto_select
class UserDTO(DefineSubset):
__subset__ = (User, ("id", "name"))
class TaskDTO(DefineSubset):
__subset__ = (Task, ("id", "title", "owner_id"))
owner: UserDTO | None = None # auto-loaded
class SprintDTO(DefineSubset):
__subset__ = (Sprint, ("id", "name"))
tasks: list[TaskDTO] = [] # auto-loaded
er = ErManager(entities=[User, Sprint, Task], session_factory=async_session)
Resolver = er.create_resolver()
async def load_sprints() -> list[SprintDTO]:
stmt = build_dto_select(SprintDTO) # root columns only
async with async_session() as session:
rows = (await session.exec(stmt)).all()
dtos = [SprintDTO(**dict(r._mapping)) for r in rows]
return await Resolver().resolve(dtos) # tree filled, batched
# 1 query per relationship, zero N+1
同一个关系模型,延伸到更复杂的架构。
DataLoader 批量加载、SQL 列裁剪、窗口函数分页;total_count 只在响应请求时才计算。
post_* 计算派生字段,ExposeAs / SendTo 实现跨层数据流。
普通 Pydantic 模型作为非表图根 —— Redis、搜索、SDK 支撑的数据进同一张图。
无中心网关组合多个 nexusx 数据图 —— 同构联邦,只组 nexusx 服务。
ComposedErManager 单进程组合多引擎;DTO 联邦跨服务加载 public DTO 树。
独立打包的应用与数据库,合并为一个 MCP 服务。
决定每个 API 形态的设计原则。
一次字段选择同时决定:GraphQL 响应形状、SQL 加载的列、拷贝进 DTO 的字段、MCP 返回、CLI --select,以及 total_count 是否计算。
Redis、搜索引擎、其他数据库、外部 API —— 声明一个带异步批量函数的 Relationship,即可加入同一套 loader、DTO、GraphQL 与 ER 图基础设施。
业务方法不依赖任何协议对象 —— 构建器检查类型化签名后挂上 REST / MCP / CLI / SDK 适配器。FromContext 注入可信值(用户、租户),不暴露为客户端参数。
把 4-phase skill 装进你的编码 Agent(Claude Code、Codex、Cursor 等),用自然语言描述你的应用。Agent 驱动流程,你只需审视模型。
npx skills add KLR-Pattern/nexusx -s nexusx-4phase -a claude-code
先和你确认领域模型与持久化策略,再动代码。
实体与关系、GraphQL 辅助接口、UseCase 的 REST / MCP / CLI 交付。
可选从 compose schema 生成类型化 TypeScript SDK。
与你现有的框架和工具无缝集成。