diff --git a/.changeset/mcp-lens-tools.md b/.changeset/mcp-lens-tools.md new file mode 100644 index 00000000..cd45bc02 --- /dev/null +++ b/.changeset/mcp-lens-tools.md @@ -0,0 +1,16 @@ +--- +'lens': minor +--- + +Add the `lens_list_pairs`, `lens_get_pools` and `lens_screen_pairs` MCP tools +(#409), so an agent that can already ask for a price can also ask what is +quotable, what liquidity stands behind it and what is moving — the three +questions REST answered and MCP did not, which left an agent guessing asset +codes. Each tool runs the function its REST route runs (`listPairs`, `listPools` +and the screener service behind `GET /screener` and the GraphQL query), so a +page over MCP is the same rows, filters and cursor as a page over HTTP. Pages +are bounded by the server's configured page limit and carry `nextCursor` and +`hasMore`; a page too large for the tool output limit is returned labelled as +truncated, without a cursor that would skip the rows it dropped. Every asset +carries its issuer in full, since a Stellar asset code alone identifies nothing, +and each description says which of the three questions it answers. diff --git a/README.md b/README.md index a141d00f..a475ce5e 100644 --- a/README.md +++ b/README.md @@ -410,8 +410,11 @@ When multiple venues quote the same pair, their prices naturally diverge due to | `paid_call` | Fetches an x402-gated URL and handles the `402 → sign → retry` loop. | **Yes — it signs with a real Stellar key** | | `payment_budget` | Reports the configured limits and what is left of today's budget. | No | | `lens_get_price` | Reads Lens's own aggregated price for a watched pair. | No | +| `lens_list_pairs` | Lists the pairs watched on one network, one bounded page at a time, each with its latest price and the issuer of both assets in full. | No | +| `lens_get_pools` | Lists AMM pools with their newest reserves, spot price and fee — how much liquidity stands behind a pair. | No | +| `lens_screen_pairs` | Ranks watched pairs by volume, 24h change, price or liquidity, with the same filters as `GET /screener`. | No | -`payment_budget` appears whenever a spend budget is configured, which the example always sets, and `lens_get_price` only where the server can reach Lens's own price data. Every one of them is served over **either** transport, and the tool list, schemas and error codes are identical whichever you pick. Which transport runs is selected by configuration; **stdio is the default**, so an existing setup needs no changes. +`payment_budget` appears whenever a spend budget is configured, which the example always sets, and `lens_get_price` only where the server can reach Lens's own price data. The three data tools follow the same rule: they are advertised only where the host supplies Lens's data, and each one runs the very function its REST route runs, so a page over MCP is the same rows and cursor as a page over HTTP. Every one of them is served over **either** transport, and the tool list, schemas and error codes are identical whichever you pick. Which transport runs is selected by configuration; **stdio is the default**, so an existing setup needs no changes. **⚠️ Warning:** the mainnet key signs with real XLM. Read the configuration, spending warning, and worked example before running. diff --git a/docs/mcp-resources.md b/docs/mcp-resources.md index 8e906e76..ba0a9c89 100644 --- a/docs/mcp-resources.md +++ b/docs/mcp-resources.md @@ -3,7 +3,8 @@ The Lens MCP server (`src/mcp/server.ts`) offers an agent two kinds of thing: - **Tools** — functions it calls to *do* something: `bazaar_search`, - `paid_call`, `lens_get_price`. + `paid_call`, `lens_get_price`, and where Lens's own data is reachable the + three data tools `lens_list_pairs`, `lens_get_pools` and `lens_screen_pairs`. - **Resources** — read-only content it attaches to its context once, instead of re-fetching by calling a tool every turn. @@ -62,7 +63,10 @@ for the same reason `createLensPriceQueries` lives in its own module: a process that has not opted into Lens's own price data should not pay for importing `pg` and Prisma just to list what the service watches. `lens://pairs` therefore reports the *declared* pair list for the network, not the process-wide runtime -registry, matching what `lens_get_price` accepts. +registry, matching what `lens_get_price` accepts. The `lens_list_pairs` tool is +the other half of that: it runs `GET /pairs`'s own listing, including pairs +added at runtime, one bounded page with a cursor at a time — which is why the +resource stays a cheap, cacheable snapshot instead of a paginated read. A host that does hold that state can inject its own descriptors, exactly as it injects `priceQueries`: diff --git a/src/__tests__/pagination.test.ts b/src/__tests__/pagination.test.ts index 101c5918..f9e5cba1 100644 --- a/src/__tests__/pagination.test.ts +++ b/src/__tests__/pagination.test.ts @@ -39,7 +39,7 @@ import { POOLS_DEFAULT_LIMIT, POOLS_MAX_LIMIT, } from '../pagination' -import { registerRESTRoutes } from '../api/rest' +import { registerPoolsRoutes } from '../routes/pools' import { registerPairsRoutes } from '../routes/pairs' describe('parseLimit', () => { @@ -111,7 +111,7 @@ describe('slicePage', () => { async function buildPoolsApp() { const app = Fastify({ logger: false }) - await registerRESTRoutes(app) + await registerPoolsRoutes(app) await app.ready() return app } diff --git a/src/__tests__/pools.test.ts b/src/__tests__/pools.test.ts index 0cab37e0..636e7e28 100644 --- a/src/__tests__/pools.test.ts +++ b/src/__tests__/pools.test.ts @@ -7,16 +7,7 @@ vi.mock('../db', () => ({ pgPool: { query: mockQuery }, })) -vi.mock('../redis', () => ({ - getCachedPrice: vi.fn(), - setCachedPrice: vi.fn(), -})) - -vi.mock('../aggregator/bestRoute', () => ({ - getBestRoute: vi.fn(), -})) - -import { registerRESTRoutes } from '../api/rest' +import { registerPoolsRoutes } from '../routes/pools' import { activeNetwork } from '../config' async function buildApp(network?: 'testnet' | 'mainnet') { @@ -27,7 +18,7 @@ async function buildApp(network?: 'testnet' | 'mainnet') { req.network = network }) } - await registerRESTRoutes(app) + await registerPoolsRoutes(app) await app.ready() return app } diff --git a/src/api/rest.ts b/src/api/rest.ts index 90e3983a..b6d37702 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -17,21 +17,12 @@ import { priceResponseSchema, routeResponseSchema, historyResponseSchema, - poolsResponseSchema, depthResponseSchema, impactResponseSchema, installResponseValidation, } from './schemas' import { findOverlappingGaps } from '../backfill/gaps' import { getDepth, getImpact } from '../pricing/depth' -import { - POOLS_DEFAULT_LIMIT, - POOLS_MAX_LIMIT, - decodeCursor, - encodeCursor, - parseLimit, - slicePage, -} from '../pagination' function makePairKey(a: string, b: string): string { return [a, b].sort().join('/') @@ -381,94 +372,9 @@ export async function registerRESTRoutes(app: FastifyInstance) { } ) - // GET /pools?cursor=&limit= - // - // Latest snapshot per pool for one network, without reading history. The pool - // ids are discovered with a recursive "skip scan" over the - // (network, pool_id, timestamp DESC) index (jump to the next distinct pool_id - // each step), then one LIMIT 1 index probe per pool fetches its newest row. - // Cost is O(pools x log n); the old DISTINCT ON sorted every row of every - // network. A quiet pool is kept with its real `timestamp`, never dropped. - // - // Paginated keyset-style on pool_id ASC (default limit 50, max 200): the - // cursor carries the last page's pool_id, and the skip scan's seed query - // resumes strictly after it, so a pool inserted mid-paging cannot shift a - // page boundary the way an OFFSET would. - app.get<{ Querystring: { cursor?: string; limit?: string } }>( - '/pools', - { config: { scope: 'read:prices' }, schema: { response: { 200: poolsResponseSchema } } }, - async (req, reply) => { - const network = req.network ?? activeNetwork - const decimalFormat = resolveDecimalFormat(req) - if (decimalFormat === 'string') { - reply.header('x-decimal-format', 'string') - } - - const parsedLimit = parseLimit(req.query.limit, POOLS_DEFAULT_LIMIT, POOLS_MAX_LIMIT) - if ('error' in parsedLimit) { - return reply.status(400).send({ error: parsedLimit.error }) - } - const { limit } = parsedLimit - - const params: unknown[] = [network] - let seedFilter = '' - if (req.query.cursor) { - const decoded = decodeCursor(req.query.cursor) - if (decoded === null) { - return reply.status(400).send({ error: 'Invalid cursor format' }) - } - if (decoded !== 'skip') { - params.push(decoded.value) - seedFilter = ` AND pool_id > $${params.length}` - } - } - params.push(limit + 1) // Fetch one extra to determine hasMore - - const result = await pgPool.query( - `WITH RECURSIVE ids AS ( - (SELECT pool_id FROM pool_snapshots WHERE network = $1${seedFilter} ORDER BY pool_id LIMIT 1) - UNION ALL - SELECT (SELECT s.pool_id FROM pool_snapshots s - WHERE s.network = $1 AND s.pool_id > ids.pool_id - ORDER BY s.pool_id LIMIT 1) - FROM ids - WHERE ids.pool_id IS NOT NULL - ) - SELECT l.pool_id, l.asset_a, l.asset_b, - l.reserve_a::numeric, l.reserve_b::numeric, l.spot_price::numeric, l.fee_bp, l.timestamp - FROM ids - CROSS JOIN LATERAL ( - SELECT * FROM pool_snapshots s - WHERE s.network = $1 AND s.pool_id = ids.pool_id - ORDER BY s.timestamp DESC - LIMIT 1 - ) l - WHERE ids.pool_id IS NOT NULL - ORDER BY l.pool_id ASC - LIMIT $${params.length}`, - params - ) - - const { data, hasMore } = slicePage(result.rows, limit) - const nextCursor = - hasMore && data.length > 0 ? encodeCursor(data[data.length - 1].pool_id, data[data.length - 1].pool_id) : null - - return { - pools: data.map(r => ({ - pool_id: r.pool_id, - asset_a: r.asset_a, - asset_b: r.asset_b, - reserve_a: formatDecimal(r.reserve_a, decimalFormat), - reserve_b: formatDecimal(r.reserve_b, decimalFormat), - spot_price: formatDecimal(r.spot_price, decimalFormat), - fee_bp: r.fee_bp !== null && r.fee_bp !== undefined ? parseInt(String(r.fee_bp), 10) : null, - timestamp: r.timestamp, - })), - nextCursor, - hasMore, - } - } - ) + // GET /pools?cursor=&limit= lives in src/routes/pools.ts, next to /pairs and + // /screener, so the `lens_get_pools` MCP tool runs the same listPools this + // route does; registerPoolsRoutes() is called from app.ts. // GET /price/:assetA/:assetB/depth?amount=1000 app.get<{ diff --git a/src/app.ts b/src/app.ts index 85de28aa..4233a362 100644 --- a/src/app.ts +++ b/src/app.ts @@ -9,6 +9,7 @@ import { registerGraphQL } from './api/graphql' import { registerWebhookRoutes } from './routes/webhooks' import { registerCandleRoutes } from './routes/candles' import { registerPairsRoutes } from './routes/pairs' +import { registerPoolsRoutes } from './routes/pools' import { registerScreenerRoutes } from './routes/screener' import { registerMoversRoutes } from './routes/movers' import { registerHistoryRoutes } from './api/history' @@ -240,6 +241,7 @@ export async function buildApp(options: BuildAppOptions = {}): Promise + listPools(network: NetworkName, page: { limit?: number; cursor?: string }): Promise + /** Pairs watched on that network, issuers intact — what a pool row's bare codes are resolved against. */ + watchedPairs(network: NetworkName): readonly { assetA: AssetId; assetB: AssetId }[] + /** + * The screener. Rejects are returned rather than thrown, so an unsupported + * filter or an undecodable cursor reaches the agent as the same message the + * REST route answers with; a database that is unreachable still throws. + * `limit` comes back as the page size actually run — the screener bounds its + * own pages, tighter than the tool ceiling, and the agent is told which one + * it got. + */ + screen(request: { + network: NetworkName + filters?: ScreenerFilters + sortBy?: string + order?: string + cursor?: string + limit?: number + /** Present at all means the caller asked for market cap, which cannot be computed. */ + marketCapAsked: boolean + }): Promise +} + +export type ScreenerScreenOutcome = + | { ok: true; page: ScreenerPage; limit: number } + | { ok: false; error: string } + +export type LensPairEntry = { + pairKey: string + assetA: AssetRef + assetB: AssetRef + latestPrice: number | null + lastUpdated: string | null + /** The venue's own observation time, when the source carries one (#243). */ + sourceTime: string | null +} + +export type LensPairsPage = { + network: NetworkName + pairs: LensPairEntry[] + /** The page size actually honoured — clamped when the ask exceeded the ceiling. */ + limit: number + nextCursor: string | null + hasMore: boolean +} + +export type LensPoolEntry = { + pool_id: string + /** The codes the AMM stores on the row, unchanged. */ + asset_a: string + asset_b: string + /** The same assets with the issuer that defines them, or null when this network watches no such pair. */ + assetA: AssetRef + assetB: AssetRef + reserve_a: number | null + reserve_b: number | null + spot_price: number | null + fee_bp: number | null + timestamp: string +} + +export type LensPoolsPage = { + network: NetworkName + pools: LensPoolEntry[] + limit: number + nextCursor: string | null + hasMore: boolean +} + +export type LensScreenerRow = { + pair_key: string + volume: number + price: number + change_24h: number + liquidity: number + liquidity_unit: string +} + +export type LensScreenerPage = { + network: NetworkName + data: LensScreenerRow[] + /** The page size the screener actually ran, which can be below the asked-for limit. */ + limit: number + nextCursor: string | null + hasMore: boolean + /** Always null: the screener ranks a window, it does not count a population. */ + total: null +} + +export type LensDataToolFailure = { code: McpErrorCode; reason: string } + +export type LensDataToolOutcome = + | { ok: true; result: LensPairsPage | LensPoolsPage | LensScreenerPage } + | { ok: false; error: LensDataToolFailure } + +const VALID_NETWORKS: readonly NetworkName[] = ['testnet', 'mainnet'] + +function assetRef(asset: AssetId): AssetRef { + return { code: asset.code, issuer: asset.issuer ?? null } +} + +function pairEntry(p: PairWithLatest): LensPairEntry { + return { + pairKey: p.pairKey, + assetA: assetRef(p.assetA), + assetB: assetRef(p.assetB), + latestPrice: p.latestPrice, + // Dates do not survive JSON; an agent needs to read the age, not infer it. + lastUpdated: p.lastUpdated ? new Date(p.lastUpdated).toISOString() : null, + sourceTime: p.sourceTime ? new Date(p.sourceTime).toISOString() : null, + } +} + +const pagingProperties = { + network: { + type: 'string', + enum: [...VALID_NETWORKS], + description: 'Stellar network to read. Defaults to the network this server is configured for.', + }, + limit: { + type: 'integer', + minimum: 1, + description: + 'Items to return in this page. Clamped to the server ceiling rather than refused; check "limit" in the result for the size you actually got.', + }, + cursor: { + type: 'string', + description: + 'Opaque token from a previous result\'s "nextCursor". Do not construct or guess one, and do not treat it as an offset.', + }, +} as const + +const pageFields = { + limit: { type: 'integer', minimum: 1 }, + nextCursor: { type: ['string', 'null'] }, + hasMore: { type: 'boolean' }, +} as const + +function assetRefSchema() { + return { + type: 'object', + properties: { + code: { type: 'string' }, + issuer: { type: ['string', 'null'] }, + }, + required: ['code', 'issuer'], + additionalProperties: false, + } +} + +const PAGING_NOTICE = + 'Results are one page: when "hasMore" is true, pass "nextCursor" back as the cursor argument for the rest. ' + +const ISSUER_NOTICE = + 'Every asset is carried as {code, issuer} with the issuer spelled out in full — a Stellar asset code ' + + 'is not unique, anyone can issue "USDC", so an issuer is part of the identity, never an optional detail. ' + +export function lensDataToolDefinitions() { + return [ + { + name: LENS_LIST_PAIRS_TOOL_NAME, + description: + 'List the asset pairs this Lens deployment watches, one page at a time, with the latest price and ' + + 'its age for each. This is the discovery tool: call it before lens_get_price to learn what can ' + + 'actually be quoted, and use the pair keys and full issuers it returns verbatim in that call. ' + + 'It answers "what exists here" — not "which pair is moving" (that is lens_screen_pairs, which ' + + 'filters and ranks) and not "what liquidity is behind a pair" (that is lens_get_pools). ' + + PAGING_NOTICE + + ISSUER_NOTICE + + 'A pair with no data on this network comes back with a null price and null lastUpdated rather than ' + + 'being dropped, so an empty listing means nothing is watched on that network.', + inputSchema: zodToToolInputSchema(LENS_LIST_PAIRS_ARGUMENTS_SCHEMA), + outputSchema: { + type: 'object', + properties: { + network: { type: 'string', enum: [...VALID_NETWORKS] }, + pairs: { + type: 'array', + items: { + type: 'object', + properties: { + pairKey: { type: 'string' }, + assetA: assetRefSchema(), + assetB: assetRefSchema(), + latestPrice: { type: ['number', 'null'] }, + lastUpdated: { type: ['string', 'null'] }, + sourceTime: { type: ['string', 'null'] }, + }, + required: ['pairKey', 'assetA', 'assetB', 'latestPrice', 'lastUpdated', 'sourceTime'], + additionalProperties: false, + }, + }, + ...pageFields, + }, + required: ['network', 'pairs', 'limit', 'nextCursor', 'hasMore'], + additionalProperties: false, + }, + }, + { + name: LENS_GET_POOLS_TOOL_NAME, + description: + 'List the AMM liquidity pools Lens ingests on one network, one page at a time, with each pool\'s ' + + 'newest reserves, spot price and fee. Use it to judge whether a price is deep enough to act on — ' + + 'how much liquidity stands behind a pair — after lens_list_pairs has told you the pair exists. ' + + 'It is not a discovery tool for pairs (use lens_list_pairs) and not a market mover list (use ' + + 'lens_screen_pairs). ' + + PAGING_NOTICE + + ISSUER_NOTICE + + 'The AMM stores only asset codes on a pool row, so the issuers attached here are the ones from the ' + + 'watched pair that carries those codes; when no watched pair does, issuer is null rather than a ' + + 'guessed one. Pages are ordered by pool id.', + inputSchema: zodToToolInputSchema(LENS_GET_POOLS_ARGUMENTS_SCHEMA), + outputSchema: { + type: 'object', + properties: { + network: { type: 'string', enum: [...VALID_NETWORKS] }, + pools: { + type: 'array', + items: { + type: 'object', + properties: { + pool_id: { type: 'string' }, + asset_a: { type: 'string' }, + asset_b: { type: 'string' }, + assetA: assetRefSchema(), + assetB: assetRefSchema(), + reserve_a: { type: ['number', 'null'] }, + reserve_b: { type: ['number', 'null'] }, + spot_price: { type: ['number', 'null'] }, + fee_bp: { type: ['integer', 'null'] }, + timestamp: { type: 'string' }, + }, + required: [ + 'pool_id', 'asset_a', 'asset_b', 'assetA', 'assetB', + 'reserve_a', 'reserve_b', 'spot_price', 'fee_bp', 'timestamp', + ], + additionalProperties: false, + }, + }, + ...pageFields, + }, + required: ['network', 'pools', 'limit', 'nextCursor', 'hasMore'], + additionalProperties: false, + }, + }, + { + name: LENS_SCREEN_PAIRS_TOOL_NAME, + description: + 'Rank watched pairs by 24h activity and filter them by volume, price, change or liquidity: the ' + + '"what is moving" view. Use it to shortlist candidates from a population, then lens_get_price on a ' + + 'specific pair for the quote and lens_get_pools for the depth behind it. Do not use it to enumerate ' + + 'what exists (that is lens_list_pairs, which is unfiltered and lists quiet pairs too). ' + + 'sortBy is volume, change_24h, price or liquidity (default volume); order is asc or desc ' + + '(default desc). market_cap is refused: Lens has no circulating-supply data to compute it from, so ' + + 'filter liquidity instead. ' + + PAGING_NOTICE + + 'Rows key on pair_key, which already names each asset\'s issuer in full, so a screen result can be ' + + 'passed straight to lens_get_price without rebuilding the asset.', + inputSchema: zodToToolInputSchema(LENS_SCREEN_PAIRS_ARGUMENTS_SCHEMA), + outputSchema: { + type: 'object', + properties: { + network: { type: 'string', enum: [...VALID_NETWORKS] }, + data: { + type: 'array', + items: { + type: 'object', + properties: { + pair_key: { type: 'string' }, + volume: { type: 'number' }, + price: { type: 'number' }, + change_24h: { type: 'number' }, + liquidity: { type: 'number' }, + liquidity_unit: { type: 'string' }, + }, + required: ['pair_key', 'volume', 'price', 'change_24h', 'liquidity', 'liquidity_unit'], + additionalProperties: false, + }, + }, + ...pageFields, + total: { type: 'null' }, + }, + required: ['network', 'data', 'limit', 'nextCursor', 'hasMore', 'total'], + additionalProperties: false, + }, + }, + ] +} + +function failure(code: McpErrorCode, reason: string): LensDataToolOutcome { + return { ok: false, error: { code, reason } } +} + +/** + * The server hands the caller's object straight to the handler, so a bad + * `network` or an argument type the schema never saw must be settled here rather + * than reach the database and come back as something worse. + */ +function readNetwork(raw: unknown, queries: LensDataQueries): NetworkName | LensDataToolFailure { + if (raw === undefined) return queries.activeNetwork + if (typeof raw !== 'string' || !(VALID_NETWORKS as readonly string[]).includes(raw as NetworkName)) { + return { + code: MCP_ERROR_CODES.ERR_BAD_REQUEST, + reason: `network must be one of ${VALID_NETWORKS.join(' or ')}, got ${JSON.stringify(raw)}`, + } + } + return raw as NetworkName +} + +function isFailure(value: NetworkName | LensDataToolFailure): value is LensDataToolFailure { + return typeof value === 'object' && value !== null && 'code' in value +} + +/** + * Clamp rather than refuse: an agent that asks for 500 pairs gets a small page + * and a cursor, which is a better outcome than an error it has to understand. + */ +function resolveLimit(limit: number | undefined, paging: LensToolPaging): number { + return limit !== undefined ? Math.min(limit, paging.maxLimit) : paging.defaultLimit +} + +export async function readLensPairsPage( + args: LensListPairsArguments, + queries: LensDataQueries, + paging: LensToolPaging +): Promise { + const network = readNetwork(args.network, queries) + if (isFailure(network)) return failure(network.code, network.reason) + + const limit = resolveLimit(args.limit, paging) + const outcome = await queries.listPairs(network, { limit, cursor: args.cursor }) + if (!outcome.ok) return failure(MCP_ERROR_CODES.ERR_BAD_REQUEST, outcome.error) + + return { + ok: true, + result: { + network, + pairs: outcome.pairs.map(pairEntry), + // The page size the listing actually honoured, which is the clamped one + // when the ask was over the ceiling. + limit: outcome.limit, + nextCursor: outcome.nextCursor, + hasMore: outcome.hasMore, + }, + } +} + +/** + * Attach the issuer a pool row does not store. The AMM snapshot carries bare + * codes; the watched pair that names those codes is what says which USDC this is. + * A pool whose codes match no watched pair keeps null issuers rather than + * borrowing the only one on hand — a wrong issuer is a different asset. + */ +export function resolvePoolAssets( + row: Pick, + pairs: readonly { assetA: AssetId; assetB: AssetId }[] +): { assetA: AssetRef; assetB: AssetRef } { + const match = pairs.find( + p => + (p.assetA.code === row.asset_a && p.assetB.code === row.asset_b) || + (p.assetA.code === row.asset_b && p.assetB.code === row.asset_a) + ) + + if (!match) { + return { + assetA: { code: row.asset_a, issuer: null }, + assetB: { code: row.asset_b, issuer: null }, + } + } + + const forward = match.assetA.code === row.asset_a + return { + assetA: assetRef(forward ? match.assetA : match.assetB), + assetB: assetRef(forward ? match.assetB : match.assetA), + } +} + +export async function readLensPoolsPage( + args: LensGetPoolsArguments, + queries: LensDataQueries, + paging: LensToolPaging +): Promise { + const network = readNetwork(args.network, queries) + if (isFailure(network)) return failure(network.code, network.reason) + + const limit = resolveLimit(args.limit, paging) + const outcome = await queries.listPools(network, { limit, cursor: args.cursor }) + if (!outcome.ok) return failure(MCP_ERROR_CODES.ERR_BAD_REQUEST, outcome.error) + + return { + ok: true, + result: { + network, + pools: outcome.pools.map(r => ({ + pool_id: r.pool_id, + asset_a: r.asset_a, + asset_b: r.asset_b, + ...resolvePoolAssets(r, queries.watchedPairs(network)), + // numeric columns arrive as strings to keep them exact; an agent reading + // JSON wants the number, and this is where the rounding is decided. + reserve_a: r.reserve_a === null ? null : parseFloat(r.reserve_a), + reserve_b: r.reserve_b === null ? null : parseFloat(r.reserve_b), + spot_price: r.spot_price === null ? null : parseFloat(r.spot_price), + fee_bp: r.fee_bp === null || r.fee_bp === undefined ? null : parseInt(String(r.fee_bp), 10), + timestamp: new Date(r.timestamp).toISOString(), + })), + limit: outcome.limit, + nextCursor: outcome.nextCursor, + hasMore: outcome.hasMore, + }, + } +} + +const SCREENER_FILTER_KEYS = [ + 'volume', + 'change_24h_min', + 'change_24h_max', + 'price_min', + 'price_max', + 'liquidity', +] as const + +export async function readLensScreenerPage( + args: LensScreenPairsArguments, + queries: LensDataQueries, + paging: LensToolPaging +): Promise { + const network = readNetwork(args.network, queries) + if (isFailure(network)) return failure(network.code, network.reason) + + const limit = resolveLimit(args.limit, paging) + + const filters: ScreenerFilters = {} + for (const key of SCREENER_FILTER_KEYS) { + const value = args[key] + if (value !== undefined) filters[key] = value + } + + const outcome = await queries.screen({ + network, + filters, + sortBy: args.sortBy, + order: args.order, + cursor: args.cursor, + limit, + // sortBy is enum-validated by the schema, so a market_cap sort is refused as + // an invalid argument; only the filter itself needs the dedicated message. + marketCapAsked: args.market_cap !== undefined, + }) + if (!outcome.ok) return failure(MCP_ERROR_CODES.ERR_BAD_REQUEST, outcome.error) + + const { page } = outcome + return { + ok: true, + result: { + network, + data: page.data.map((row: ScreenerRow & { liquidity_unit: string }) => ({ + pair_key: row.pair_key, + volume: row.volume, + price: row.price, + change_24h: row.change_24h, + liquidity: row.liquidity, + liquidity_unit: row.liquidity_unit, + })), + limit: outcome.limit, + nextCursor: page.nextCursor, + hasMore: page.hasMore, + total: page.total, + }, + } +} + +/** Run one of the three Lens data tools. The caller has already matched the name. */ +export async function callLensDataTool( + name: LensDataToolName, + args: unknown, + queries: LensDataQueries, + paging: LensToolPaging +): Promise { + const schema = + name === LENS_LIST_PAIRS_TOOL_NAME + ? LENS_LIST_PAIRS_ARGUMENTS_SCHEMA + : name === LENS_GET_POOLS_TOOL_NAME + ? LENS_GET_POOLS_ARGUMENTS_SCHEMA + : LENS_SCREEN_PAIRS_ARGUMENTS_SCHEMA + + const parsed = schema.safeParse(args ?? {}) + if (!parsed.success) { + // Naming the offending argument is the difference between an agent fixing + // its call and an agent retrying the same mistake. + return failure(MCP_ERROR_CODES.ERR_BAD_REQUEST, formatZodError(parsed.error)) + } + + switch (name) { + case LENS_LIST_PAIRS_TOOL_NAME: + return readLensPairsPage(parsed.data as LensListPairsArguments, queries, paging) + case LENS_GET_POOLS_TOOL_NAME: + return readLensPoolsPage(parsed.data as LensGetPoolsArguments, queries, paging) + case LENS_SCREEN_PAIRS_TOOL_NAME: + return readLensScreenerPage(parsed.data as LensScreenPairsArguments, queries, paging) + } +} diff --git a/src/mcp/server.ts b/src/mcp/server.ts index 320a03a5..3e5051fb 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -29,6 +29,15 @@ import { PaymentLedger } from './ledger.js'; import { getMetrics, register } from '../metrics.js'; import { checkPaidCallTarget, parsePaidCallAllowlist, type PaidCallAllowRule } from './paidCallAllowlist.js'; import { LENS_PRICE_TOOL_NAME, lensPriceToolDefinition, readLensPrice, type LensPriceQueries } from './priceTool.js'; +import { + LENS_GET_POOLS_TOOL_NAME, + LENS_LIST_PAIRS_TOOL_NAME, + LENS_SCREEN_PAIRS_TOOL_NAME, + callLensDataTool, + isLensDataToolName, + lensDataToolDefinitions, + type LensDataQueries, +} from './lensTools.js'; import { BAZAAR_SEARCH_ARGUMENTS_SCHEMA, PAID_CALL_ARGUMENTS_SCHEMA, @@ -118,6 +127,13 @@ export interface McpServerConfig { * not advertise the tool, so an agent is never offered a price it cannot get. */ priceQueries?: LensPriceQueries; + /** + * Lens's own pairs, pools and screener, for `lens_list_pairs`, + * `lens_get_pools` and `lens_screen_pairs` (#409). Hosts that can read the + * data pass {@link createLensDataQueries}; leaving it out does not advertise + * the tools, so an agent is never offered a listing this server cannot produce. + */ + dataQueries?: LensDataQueries; /** * Outbound fetch policy for `bazaar_search`. Falls back to the operator's * shared `OUTBOUND_ALLOWED_HOSTS` allowlist. `lookup`/`fetchImpl` exist so @@ -273,6 +289,7 @@ export class BazaarMcpServer { private paidCallRules: PaidCallAllowRule[]; private bazaarFetchOptions: SsrfFetchOptions; private priceQueries?: LensPriceQueries; + private dataQueries?: LensDataQueries; private budget?: SpendBudget; private limits: McpLimits; private resources: LensResourceDescriptor[]; @@ -282,6 +299,7 @@ export class BazaarMcpServer { constructor(config_: McpServerConfig) { this.fetchWithPayment = config_.fetchWithPayment; this.priceQueries = config_.priceQueries; + this.dataQueries = config_.dataQueries; this.budget = config_.budget; this.paymentLedger = config_.paymentLedger ?? new PaymentLedger(); this.limits = { ...config.mcp, ...config_.limits }; @@ -375,6 +393,13 @@ export class BazaarMcpServer { tools.push(lensPriceToolDefinition()); } + // #409: the same reasoning one question later — an agent that can read a + // price still cannot ask what is quotable, what liquidity backs it, or what + // is moving. Advertised only when this process can read Lens's data. + if (this.dataQueries) { + tools.push(...lensDataToolDefinitions()); + } + return { tools }; }); @@ -478,6 +503,58 @@ export class BazaarMcpServer { } } + if (isLensDataToolName(request.params.name)) { + if (!this.dataQueries) { + return this.createErrorResponse( + 'ERR_INTERNAL_FAILURE', + `${request.params.name} is not available: this server has no Lens data source configured.` + ); + } + + try { + const outcome = await callLensDataTool(request.params.name, request.params.arguments, this.dataQueries, { + defaultLimit: this.limits.defaultPageLimit, + maxLimit: this.limits.maxPageLimit, + }); + + if (!outcome.ok) { + return this.createErrorResponse(outcome.error.code, outcome.error.reason); + } + + // Same rule as the price tool: a page that does not fit the output + // budget is reported as truncated rather than shipped whole, because + // the point of #409 is that an agent is not handed 500 pairs at once. + // The cursor goes first: it names the end of the page that was + // fetched, not of the prefix that survives, so following it would + // skip rows silently. `hasMore` plus the notice tell the agent to + // re-run the same call with a smaller limit instead. + const budget: ResultBudget = { + maxBytes: this.limits.maxResultBytes, + maxItems: Number.MAX_SAFE_INTEGER, + offset: 0, + notice: `A Lens data page was truncated to fit a ${this.limits.maxResultBytes}-byte tool output limit; ask again with a smaller limit for the rest.`, + }; + const page = + Buffer.byteLength(JSON.stringify(outcome.result), 'utf8') > budget.maxBytes + ? { ...outcome.result, nextCursor: null } + : outcome.result; + const bounded = boundSerialized(page, budget); + return bounded.truncated + ? { content: [{ type: 'text', text: bounded.text }] } + : { content: [{ type: 'text', text: bounded.text }], structuredContent: outcome.result }; + } catch (e: any) { + // These pages come from Postgres, so a throw means the data layer is + // unreachable. The reason says that and nothing more: a driver error + // carries the connection string, and an agent's transcript is not where + // it belongs. The operator sees the real failure in the log. + console.error(`[mcp-server] ${request.params.name} failed:`, e); + return this.createErrorResponse( + 'ERR_SERVICE_UNREACHABLE', + `${request.params.name} could not read Lens data: the backing store is unreachable. Retry later rather than assuming the answer is empty.` + ); + } + } + if (request.params.name === 'bazaar_search') { const parsed = BAZAAR_SEARCH_ARGUMENTS_SCHEMA.safeParse(request.params.arguments ?? {}); if (!parsed.success) { diff --git a/src/mcp/toolSchemas.ts b/src/mcp/toolSchemas.ts index 89c02cd0..d156de33 100644 --- a/src/mcp/toolSchemas.ts +++ b/src/mcp/toolSchemas.ts @@ -64,6 +64,69 @@ export const PAID_CALL_ARGUMENTS_SCHEMA = z export type PaidCallArguments = z.infer; +/** + * #409 — the paging half of the Lens data tools. `limit` is deliberately not + * capped here: an over-large ask is clamped to the operator's + * `MCP_MAX_PAGE_LIMIT` the same way `bazaar_search` clamps it, so the agent gets + * a smaller page rather than a refusal it has to decode. `cursor` is the opaque + * `nextCursor` the previous page handed back — REST's screener pages by the same + * token, which is what keeps a tool page and an HTTP page identical. + */ +const networkArg = z + .enum(['testnet', 'mainnet']) + .optional() + .describe('Stellar network to read. Defaults to the network this server is configured for.'); +const limitArg = z + .number() + .int() + .positive() + .optional() + .describe('Maximum number of items to return in this page (clamped to the server page ceiling)'); +const cursorArg = z + .string() + .min(1) + .optional() + .describe('Opaque cursor from a previous page. Pass back the "nextCursor" the last result gave you; it is not a number an agent should construct.'); + +export const LENS_LIST_PAIRS_ARGUMENTS_SCHEMA = z + .object({ network: networkArg, limit: limitArg, cursor: cursorArg }) + .strict(); + +export type LensListPairsArguments = z.infer; + +export const LENS_GET_POOLS_ARGUMENTS_SCHEMA = z + .object({ network: networkArg, limit: limitArg, cursor: cursorArg }) + .strict(); + +export type LensGetPoolsArguments = z.infer; + +/** + * Screener filters, one per REST query parameter. Coerced so an agent that + * sends `"1000"` instead of `1000` is understood rather than refused, which is + * also how the REST route reads them. + */ +const screenerNum = (name: string) => + z.coerce.number({ error: `${name} must be a valid number` }).optional(); + +export const LENS_SCREEN_PAIRS_ARGUMENTS_SCHEMA = z + .object({ + network: networkArg, + limit: limitArg, + cursor: cursorArg, + volume: screenerNum('volume'), + change_24h_min: screenerNum('change_24h_min'), + change_24h_max: screenerNum('change_24h_max'), + price_min: screenerNum('price_min'), + price_max: screenerNum('price_max'), + liquidity: screenerNum('liquidity'), + sortBy: z.enum(['volume', 'change_24h', 'price', 'liquidity']).optional(), + order: z.enum(['asc', 'desc']).optional(), + market_cap: z.string().optional(), + }) + .strict(); + +export type LensScreenPairsArguments = z.infer; + /** * Converts a Zod object schema into an MCP-compliant JSON schema dictionary. * Strips the top-level `$schema` field so it plugs directly into MCP tool `inputSchema`. diff --git a/src/routes/pairs.ts b/src/routes/pairs.ts index ee337217..7214f047 100644 --- a/src/routes/pairs.ts +++ b/src/routes/pairs.ts @@ -1,7 +1,8 @@ import type { FastifyInstance } from 'fastify' import { getActivePairs, parseAssetStr, makePairKey, registerPair, persistPair, hasPair } from '../pairsRegistry' import { pgPool } from '../db' -import { activeNetwork } from '../config' +import { activeNetwork, type NetworkName } from '../config' +import type { AssetId, WatchedPair } from '../types' import { PAIRS_DEFAULT_LIMIT, PAIRS_MAX_LIMIT, @@ -13,89 +14,139 @@ import { import '../middleware/network' // declares req.network on the FastifyRequest type import { admin_auth_failures_total } from '../metrics' +export interface PairWithLatest { + pairKey: string + assetA: AssetId + assetB: AssetId + latestPrice: number | null + lastUpdated: Date | null + sourceTime: Date | null +} + +export interface ListPairsPage { + /** Requested page size, clamped into [1, PAIRS_MAX_LIMIT] the way the route does. */ + limit?: number + /** Opaque cursor from a previous `nextCursor`. */ + cursor?: string +} + +export type ListPairsOutcome = + | { ok: true; pairs: PairWithLatest[]; limit: number; nextCursor: string | null; hasMore: boolean } + | { ok: false; error: string } + +/** + * The pairs listing behind GET /pairs and the `lens_list_pairs` MCP tool. + * + * One index probe per watched pair on (network, pair_key, timestamp DESC): + * `ORDER BY timestamp DESC LIMIT 1` reads a single index entry, so the cost is + * O(pairs x log n) and no longer grows with the size of the price history. + * (This used to be DISTINCT ON over the whole table, every network included.) + * + * A pair that has gone quiet is deliberately NOT dropped and NOT bounded away: + * the probe still finds its last print, and the response carries that print's + * real `lastUpdated`, so a caller can see it is old. A pair with no rows on + * this network at all comes back with null price and timestamp. + * + * Paginated keyset-style on pairKey ASC (default limit 50, max 200). The rows + * are assembled in memory from the registry, so the keyset filter and the + * limit+1 probe are applied to the merged list; the cursor carries the last + * page's pairKey, so a pair registered mid-paging cannot shift a page boundary + * the way an offset would. `limit` in the outcome is the page size actually + * honoured, which is what lets the MCP tool tell an agent it asked for 500 and + * got 200 rather than silently answering less. + * + * `watched` defaults to the process-wide runtime registry, which is what the REST + * route has always listed. A caller naming a network other than this process's + * own must pass that network's declared pairs instead, or a mainnet request + * would be answered with the pairs this testnet process watches — the same + * mistake `lens_get_price` documents in ./priceQueries.ts. + */ +export async function listPairs( + network: NetworkName, + page: ListPairsPage = {}, + watched: readonly WatchedPair[] = getActivePairs() +): Promise { + const activePairs = watched + const limit = Math.min(Math.max(page.limit ?? PAIRS_DEFAULT_LIMIT, 1), PAIRS_MAX_LIMIT) + + let afterKey: string | null = null + if (page.cursor !== undefined) { + const decoded = decodeCursor(page.cursor) + // 'skip' is a well-formed but degenerate cursor: serve the first page rather + // than punish a caller who cannot know they produced one. + if (decoded === null) return { ok: false, error: 'Invalid cursor format' } + if (decoded !== 'skip') afterKey = String(decoded.value) + } + + const result = await pgPool.query( + `SELECT p.pair_key, l.price, l.observed_at, l.source_time, l.timestamp + FROM unnest($2::text[]) AS p(pair_key) + CROSS JOIN LATERAL ( + SELECT price, COALESCE(observed_at, timestamp) AS observed_at, source_time, timestamp + FROM price_points + WHERE network = $1 + AND pair_key = p.pair_key + ORDER BY COALESCE(observed_at, timestamp) DESC + LIMIT 1 + ) l`, + [network, activePairs.map(p => p.pairKey)] + ) + + const latestPrices = new Map() + result.rows.forEach(row => { + latestPrices.set(row.pair_key, { + price: parseFloat(row.price), + observedAt: row.observed_at ?? row.timestamp, + sourceTime: row.source_time ? new Date(row.source_time) : null, + }) + }) + + const rows = activePairs + .map(p => { + const latest = latestPrices.get(p.pairKey) + return { + pairKey: p.pairKey, + assetA: p.assetA, + assetB: p.assetB, + latestPrice: latest?.price ?? null, + lastUpdated: latest?.observedAt ?? null, + sourceTime: latest?.sourceTime ?? null, + } + }) + .sort((a, b) => (a.pairKey < b.pairKey ? -1 : a.pairKey > b.pairKey ? 1 : 0)) + + const pageSource = afterKey ? rows.filter(r => r.pairKey > afterKey) : rows + const { data, hasMore } = slicePage(pageSource, limit) + const last = data[data.length - 1] + const nextCursor = hasMore && last ? encodeCursor(last.pairKey, last.pairKey) : null + return { ok: true, pairs: data, limit, nextCursor, hasMore } +} + export async function registerPairsRoutes(app: FastifyInstance) { - // GET /pairs?cursor=&limit= — list active pairs with latest price metadata - // - // One index probe per watched pair on (network, pair_key, timestamp DESC): - // `ORDER BY timestamp DESC LIMIT 1` reads a single index entry, so the cost is - // O(pairs x log n) and no longer grows with the size of the price history. - // (This used to be DISTINCT ON over the whole table, every network included.) + // GET /pairs?cursor=&limit= — list active pairs with their latest price. // - // A pair that has gone quiet is deliberately NOT dropped and NOT bounded away: - // the probe still finds its last print, and the response carries that print's - // real `lastUpdated`, so a caller can see it is old. A pair with no rows on - // this network at all comes back with null price and timestamp. - // - // Paginated keyset-style on pairKey ASC (default limit 50, max 200). The rows - // are assembled in memory from the registry, so the keyset filter and the - // limit+1 probe are applied to the merged list; the cursor carries the last - // page's pairKey, so a pair registered mid-paging cannot shift a page - // boundary the way an offset would. + // The listing itself is listPairs(), which the `lens_list_pairs` MCP tool also + // runs, so a page over MCP is the same rows and the same cursor as a page over + // HTTP. This handler only turns a query string into a page request and the + // page back into a reply: a non-numeric limit is the route's own 400, exactly + // as /screener answers one. app.get<{ Querystring: { cursor?: string; limit?: string } }>( '/pairs', { config: { scope: 'read:prices' } }, async (req, reply) => { - const activePairs = getActivePairs() - const network = req.network ?? activeNetwork - - const parsedLimit = parseLimit(req.query.limit, PAIRS_DEFAULT_LIMIT, PAIRS_MAX_LIMIT) - if ('error' in parsedLimit) { - return reply.status(400).send({ error: parsedLimit.error }) - } - const { limit } = parsedLimit - - let afterKey: string | null = null - if (req.query.cursor) { - const decoded = decodeCursor(req.query.cursor) - if (decoded === null) { - return reply.status(400).send({ error: 'Invalid cursor format' }) + const parsedLimit = parseLimit(req.query.limit, PAIRS_DEFAULT_LIMIT, PAIRS_MAX_LIMIT) + if ('error' in parsedLimit) { + return reply.status(400).send({ error: parsedLimit.error }) } - if (decoded !== 'skip') afterKey = String(decoded.value) - } - - const result = await pgPool.query( - `SELECT p.pair_key, l.price, l.observed_at, l.source_time, l.timestamp - FROM unnest($2::text[]) AS p(pair_key) - CROSS JOIN LATERAL ( - SELECT price, COALESCE(observed_at, timestamp) AS observed_at, source_time, timestamp - FROM price_points - WHERE network = $1 - AND pair_key = p.pair_key - ORDER BY COALESCE(observed_at, timestamp) DESC - LIMIT 1 - ) l`, - [network, activePairs.map(p => p.pairKey)] - ) - - const latestPrices = new Map() - result.rows.forEach(row => { - latestPrices.set(row.pair_key, { - price: parseFloat(row.price), - observedAt: row.observed_at ?? row.timestamp, - sourceTime: row.source_time ? new Date(row.source_time) : null, - }) - }) - const rows = activePairs - .map(p => { - const latest = latestPrices.get(p.pairKey) - return { - pairKey: p.pairKey, - assetA: p.assetA, - assetB: p.assetB, - latestPrice: latest?.price ?? null, - lastUpdated: latest?.observedAt ?? null, - sourceTime: latest?.sourceTime ?? null, - } + const outcome = await listPairs(req.network ?? activeNetwork, { + limit: parsedLimit.limit, + cursor: req.query.cursor, }) - .sort((a, b) => (a.pairKey < b.pairKey ? -1 : a.pairKey > b.pairKey ? 1 : 0)) - - const pageSource = afterKey ? rows.filter(r => r.pairKey > afterKey) : rows - const { data, hasMore } = slicePage(pageSource, limit) - const nextCursor = hasMore && data.length > 0 ? encodeCursor(data[data.length - 1].pairKey, data[data.length - 1].pairKey) : null - - return { pairs: data, nextCursor, hasMore } - }) + if (!outcome.ok) return reply.status(400).send({ error: outcome.error }) + return { pairs: outcome.pairs, nextCursor: outcome.nextCursor, hasMore: outcome.hasMore } + } + ) // POST /pairs — add a new trading pair at runtime app.post<{ diff --git a/src/routes/pools.ts b/src/routes/pools.ts new file mode 100644 index 00000000..da99695f --- /dev/null +++ b/src/routes/pools.ts @@ -0,0 +1,146 @@ +import type { FastifyInstance } from 'fastify' +import { pgPool } from '../db' +import { activeNetwork, type NetworkName } from '../config' +import { poolsResponseSchema } from '../api/schemas' +import { POOLS_DEFAULT_LIMIT, POOLS_MAX_LIMIT, decodeCursor, encodeCursor, parseLimit, slicePage } from '../pagination' +import { formatDecimal, resolveDecimalFormat } from '../utils/decimals' +import '../middleware/network' // declares req.network on the FastifyRequest type + +/** + * One pool's newest snapshot, straight from the DB. `numeric` columns arrive as + * strings — that is what lets `?decimalFormat=string` hand back the exact value + * the ledger stored rather than a float-rounded one — so callers decide how to + * present them via {@link formatDecimal}. + */ +export interface PoolSnapshotRow { + pool_id: string + asset_a: string + asset_b: string + reserve_a: string | null + reserve_b: string | null + spot_price: string | null + fee_bp: number | null + timestamp: Date +} + +export interface ListPoolsPage { + /** Requested page size, clamped into [1, POOLS_MAX_LIMIT] the way the route does. */ + limit?: number + /** Opaque cursor from a previous `nextCursor`. */ + cursor?: string +} + +export type ListPoolsOutcome = + | { ok: true; pools: PoolSnapshotRow[]; limit: number; nextCursor: string | null; hasMore: boolean } + | { ok: false; error: string } + +/** + * The pools listing behind GET /pools and the `lens_get_pools` MCP tool. + * + * Latest snapshot per pool for one network, without reading history. The pool + * ids are discovered with a recursive "skip scan" over the + * (network, pool_id, timestamp DESC) index (jump to the next distinct pool_id + * each step), then one LIMIT 1 index probe per pool fetches its newest row. + * Cost is O(pools x log n); the old DISTINCT ON sorted every row of every + * network. A quiet pool is kept with its real `timestamp`, never dropped. + * + * Paginated keyset-style on pool_id ASC (default limit 50, max 200): the cursor + * carries the last page's pool_id and the skip scan's seed resumes strictly + * after it, so a pool inserted mid-paging cannot shift a page boundary the way + * an offset would. `limit` in the outcome is the page size actually honoured — + * the MCP tool echoes it so an agent that asked for 500 knows it got 200. + */ +export async function listPools(network: NetworkName, page: ListPoolsPage = {}): Promise { + const limit = Math.min(Math.max(page.limit ?? POOLS_DEFAULT_LIMIT, 1), POOLS_MAX_LIMIT) + + const params: unknown[] = [network] + let seedFilter = '' + if (page.cursor !== undefined) { + const decoded = decodeCursor(page.cursor) + // 'skip' is a well-formed but degenerate cursor: serve the first page rather + // than punish a caller who cannot know they produced one. + if (decoded === null) return { ok: false, error: 'Invalid cursor format' } + if (decoded !== 'skip') { + params.push(decoded.value) + seedFilter = ` AND pool_id > $${params.length}` + } + } + params.push(limit + 1) // Fetch one extra to determine hasMore + + const result = await pgPool.query( + `WITH RECURSIVE ids AS ( + (SELECT pool_id FROM pool_snapshots WHERE network = $1${seedFilter} ORDER BY pool_id LIMIT 1) + UNION ALL + SELECT (SELECT s.pool_id FROM pool_snapshots s + WHERE s.network = $1 AND s.pool_id > ids.pool_id + ORDER BY s.pool_id LIMIT 1) + FROM ids + WHERE ids.pool_id IS NOT NULL + ) + SELECT l.pool_id, l.asset_a, l.asset_b, + l.reserve_a::numeric, l.reserve_b::numeric, l.spot_price::numeric, l.fee_bp, l.timestamp + FROM ids + CROSS JOIN LATERAL ( + SELECT * FROM pool_snapshots s + WHERE s.network = $1 AND s.pool_id = ids.pool_id + ORDER BY s.timestamp DESC + LIMIT 1 + ) l + WHERE ids.pool_id IS NOT NULL + ORDER BY l.pool_id ASC + LIMIT $${params.length}`, + params + ) + + const { data, hasMore } = slicePage(result.rows as PoolSnapshotRow[], limit) + const last = data[data.length - 1] + const nextCursor = hasMore && last ? encodeCursor(last.pool_id, last.pool_id) : null + return { ok: true, pools: data, limit, nextCursor, hasMore } +} + +export async function registerPoolsRoutes(app: FastifyInstance) { + // GET /pools?cursor=&limit= — the listing above, rendered with the caller's + // chosen decimal format. + app.get<{ Querystring: { cursor?: string; limit?: string } }>( + '/pools', + { + config: { scope: 'read:prices' }, + schema: { + response: { + 200: poolsResponseSchema, + 400: { type: 'object', required: ['error'], properties: { error: { type: 'string' } } }, + }, + }, + }, + async (req, reply) => { + const network = req.network ?? activeNetwork + const decimalFormat = resolveDecimalFormat(req) + if (decimalFormat === 'string') { + reply.header('x-decimal-format', 'string') + } + + const parsedLimit = parseLimit(req.query.limit, POOLS_DEFAULT_LIMIT, POOLS_MAX_LIMIT) + if ('error' in parsedLimit) { + return reply.status(400).send({ error: parsedLimit.error }) + } + + const outcome = await listPools(network, { limit: parsedLimit.limit, cursor: req.query.cursor }) + if (!outcome.ok) return reply.status(400).send({ error: outcome.error }) + + return { + pools: outcome.pools.map(r => ({ + pool_id: r.pool_id, + asset_a: r.asset_a, + asset_b: r.asset_b, + reserve_a: formatDecimal(r.reserve_a, decimalFormat), + reserve_b: formatDecimal(r.reserve_b, decimalFormat), + spot_price: formatDecimal(r.spot_price, decimalFormat), + fee_bp: r.fee_bp !== null && r.fee_bp !== undefined ? parseInt(String(r.fee_bp), 10) : null, + timestamp: r.timestamp, + })), + nextCursor: outcome.nextCursor, + hasMore: outcome.hasMore, + } + } + ) +} diff --git a/tests/decimalStrings.test.ts b/tests/decimalStrings.test.ts index c264fd18..41049f19 100644 --- a/tests/decimalStrings.test.ts +++ b/tests/decimalStrings.test.ts @@ -49,11 +49,15 @@ vi.mock('../src/config', () => ({ import { registerRESTRoutes } from '../src/api/rest' import { registerPriceRoutes } from '../src/routes/price' +import { registerPoolsRoutes } from '../src/routes/pools' async function buildApp() { const app = Fastify({ logger: false }) await registerRESTRoutes(app) await registerPriceRoutes(app) + // /pools lives in routes/pools.ts, which app.ts registers alongside the rest + // aggregate — a harness that wants it has to mount it too. + await registerPoolsRoutes(app) await app.ready() return app } diff --git a/tests/mcp-lens.test.ts b/tests/mcp-lens.test.ts new file mode 100644 index 00000000..f814988d --- /dev/null +++ b/tests/mcp-lens.test.ts @@ -0,0 +1,461 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import Fastify from 'fastify' +import { BazaarMcpServer } from '../src/mcp/server' +import { createLensDataQueries } from '../src/mcp/lensDataQueries' +import type { McpLimits } from '../src/mcp/limits' +import { CallToolRequestSchema, ListToolsRequestSchema } from '@modelcontextprotocol/sdk/types.js' +import { registerNetworkSelector } from '../src/middleware/network' +import { registerPairsRoutes } from '../src/routes/pairs' +import { registerPoolsRoutes } from '../src/routes/pools' +import { registerScreenerRoutes } from '../src/routes/screener' +import { MARKET_CAP_UNSUPPORTED } from '../src/services/screener' +import { activeNetwork, config, getNetworkConfig } from '../src/config' + +// The lens tools share their implementations with the REST routes, so one +// mocked database and one mocked pairs registry answer both surfaces. +const { mockQuery, mockGetActivePairs } = vi.hoisted(() => ({ + mockQuery: vi.fn(), + mockGetActivePairs: vi.fn(), +})) + +vi.mock('../src/db', () => ({ + pgPool: { query: mockQuery }, + prisma: {}, +})) + +vi.mock('../src/pairsRegistry', () => ({ + getActivePairs: mockGetActivePairs, + hasPair: vi.fn().mockReturnValue(false), + registerPair: vi.fn(), + persistPair: vi.fn(), + parseAssetStr: vi.fn(), + makePairKey: vi.fn(), +})) + +const USDC_ISSUER = 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' +const XLM = { code: 'XLM', issuer: null } +const USDC = { code: 'USDC', issuer: USDC_ISSUER } +const OLD = new Date('2026-10-16T12:00:00.000Z') + +const WATCHED = [ + { pairKey: `USDC:${USDC_ISSUER}/XLM`, assetA: XLM, assetB: USDC }, +] + +function makePairs(count: number) { + return Array.from({ length: count }, (_, i) => { + const code = `A${String(i).padStart(3, '0')}` + const issuer = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA' + return { pairKey: `${code}:${issuer}/XLM`, assetA: XLM, assetB: { code, issuer } } + }) +} + +// ── Harnesses ───────────────────────────────────────────────────────────────── + +/** + * `withData: false` builds the server as a host without Lens's own data would: + * the three tools must not be advertised at all. `limits` overrides the operator + * bounds so a test of the page ceiling is not confounded by the byte ceiling. + */ +function buildServer(withData = true, limits?: Partial) { + const server = new BazaarMcpServer({ + fetchWithPayment: vi.fn(), + ...(withData ? { dataQueries: createLensDataQueries() } : {}), + ...(limits ? { limits } : {}), + }) + const handlers: Record = {} + ;(server as any).server.setRequestHandler = (schema: any, fn: any) => { + if (schema === CallToolRequestSchema) handlers.call = fn + if (schema === ListToolsRequestSchema) handlers.list = fn + } + ;(server as any).setupHandlers() + return { + listTools: async () => (await handlers.list()).tools, + callTool: async (name: string, args?: Record) => + handlers.call({ params: { name, arguments: args } }), + } +} + +async function callToolJson(server: ReturnType, name: string, args?: Record) { + const response = await server.callTool(name, args) + expect(response.isError, `tool ${name} errored: ${response.content[0].text}`).toBeFalsy() + return response.structuredContent ?? JSON.parse(response.content[0].text) +} + +async function buildRestApp() { + const app = Fastify({ logger: false }) + await app.register(registerNetworkSelector) + await registerPairsRoutes(app) + await registerPoolsRoutes(app) + await registerScreenerRoutes(app) + await app.ready() + return app +} + +// Answers the pairs price probe with one print per queried pair key. +function mockLatestPrices() { + mockQuery.mockImplementation(async (_sql: string, params: unknown[]) => { + const keys = (params as string[][])?.[1] ?? [] + return { + rows: keys.map(pair_key => ({ pair_key, price: '0.42', timestamp: OLD, source_time: OLD })), + } + }) +} + +beforeEach(() => { + mockQuery.mockReset().mockResolvedValue({ rows: [] }) + mockGetActivePairs.mockReset().mockReturnValue([]) +}) + +// ── Tool surface ────────────────────────────────────────────────────────────── + +describe('lens MCP tools', () => { + it('lists the three lens tools next to the bazaar tools', async () => { + const server = buildServer() + const tools = await server.listTools() + const names = tools.map((t: { name: string }) => t.name) + expect(names).toEqual(expect.arrayContaining(['bazaar_search', 'paid_call'])) + expect(names).toEqual( + expect.arrayContaining(['lens_list_pairs', 'lens_get_pools', 'lens_screen_pairs']) + ) + }) + + it('does not advertise a tool it cannot answer with', async () => { + const server = buildServer(false) + const names = (await server.listTools()).map((t: { name: string }) => t.name) + expect(names).not.toContain('lens_list_pairs') + expect(names).not.toContain('lens_get_pools') + expect(names).not.toContain('lens_screen_pairs') + + // Calling one anyway refuses with the reason, rather than returning an empty + // page an agent would read as "nothing is watched here". + const response = await server.callTool('lens_list_pairs', {}) + expect(response.isError).toBe(true) + const payload = JSON.parse(response.content[0].text) + expect(payload.code).toBe('ERR_INTERNAL_FAILURE') + expect(payload.reason).toMatch(/no Lens data source/) + }) + + it('describes each tool distinctly and documents the pagination', async () => { + const server = buildServer() + const tools = (await server.listTools()) as { name: string; description: string; inputSchema: any }[] + const byName = Object.fromEntries(tools.map(t => [t.name, t])) + + expect(byName.lens_list_pairs.description).toMatch(/discover/i) + expect(byName.lens_list_pairs.description).toMatch(/lens_screen_pairs/) + expect(byName.lens_list_pairs.description).toMatch(/nextCursor/) + + expect(byName.lens_get_pools.description).toMatch(/liquidity/i) + expect(byName.lens_get_pools.description).toMatch(/lens_list_pairs/) + expect(byName.lens_get_pools.description).toMatch(/nextCursor/) + + expect(byName.lens_screen_pairs.description).toMatch(/rank/i) + expect(byName.lens_screen_pairs.description).toMatch(/lens_get_pools/) + expect(byName.lens_screen_pairs.description).toMatch(/nextCursor/) + + for (const name of ['lens_list_pairs', 'lens_get_pools', 'lens_screen_pairs']) { + // An issuer is part of an asset's identity on Stellar, so each description + // has to say the answer carries it in full. + expect(byName[name].description).toMatch(/issuer/) + expect(byName[name].inputSchema.properties.cursor).toBeTruthy() + expect(byName[name].inputSchema.properties.limit).toBeTruthy() + expect(byName[name].inputSchema.additionalProperties).toBe(false) + } + }) +}) + +// ── lens_list_pairs ─────────────────────────────────────────────────────────── + +describe('lens_list_pairs vs GET /pairs', () => { + it('returns the same pairs page by page, issuers in full', async () => { + mockGetActivePairs.mockReturnValue(makePairs(3)) + mockLatestPrices() + const server = buildServer() + const app = await buildRestApp() + + // The route's page and the tool's page for the same arguments: same rows, the + // same keyset cursor, the same verdict on whether another page exists. The + // tool adds only the page size it honoured and the network it read. + const rest1 = (await app.inject({ method: 'GET', url: '/pairs?limit=2' })).json() + const page1 = await callToolJson(server, 'lens_list_pairs', { limit: 2 }) + expect(page1).toEqual({ network: activeNetwork, limit: 2, ...rest1 }) + + // Not just equal output: the identical SQL, because the tool runs the + // route's own listing. + const restSql = mockQuery.mock.calls[0][0] + expect(mockQuery.mock.calls.at(-1)![0]).toBe(restSql) + + const rest2 = ( + await app.inject({ method: 'GET', url: `/pairs?limit=2&cursor=${encodeURIComponent(rest1.nextCursor)}` }) + ).json() + const page2 = await callToolJson(server, 'lens_list_pairs', { limit: 2, cursor: page1.nextCursor }) + expect(page2).toEqual({ network: activeNetwork, limit: 2, ...rest2 }) + expect(page2.hasMore).toBe(false) + expect(page2.nextCursor).toBe(null) + + // Issuers are complete strkeys, never shortened. + expect(page1.pairs[0].assetB.issuer).toMatch(/^G[A-Z2-7]{55}$/) + expect(page1.pairs[0].assetB.issuer).toHaveLength(56) + // And the freshness metadata REST carries travels too: an agent deciding + // whether a price is worth acting on needs the age, not just the number. + expect(page1.pairs[0].lastUpdated).toBe(OLD.toISOString()) + expect(page1.pairs[0].sourceTime).toBe(OLD.toISOString()) + await app.close() + }) + + it('bounds the page by default and clamps an oversized limit', async () => { + mockGetActivePairs.mockReturnValue(makePairs(260)) + mockLatestPrices() + // Roomy byte budget: this test is about the page ceiling, and a 200-pair + // page would otherwise be truncated by the output-size one. + const server = buildServer(true, { maxResultBytes: 4 * 1024 * 1024 }) + + const page = await callToolJson(server, 'lens_list_pairs', {}) + expect(page.pairs).toHaveLength(config.mcp.defaultPageLimit) + expect(page.hasMore).toBe(true) + expect(page.nextCursor).toBeTruthy() + + // An ask over the ceiling is clamped rather than refused, and the result says + // what size was actually honoured. + const capped = await callToolJson(server, 'lens_list_pairs', { limit: 500 }) + expect(capped.limit).toBe(config.mcp.maxPageLimit) + expect(capped.pairs).toHaveLength(config.mcp.maxPageLimit) + }) + + it('labels a page too large for the output budget instead of shipping it whole', async () => { + mockGetActivePairs.mockReturnValue(makePairs(260)) + mockLatestPrices() + const server = buildServer() + + // maxPageLimit pairs of full strkeys is far past maxResultBytes, so what the + // agent gets is a labelled fragment — and the cursor that would have skipped + // the rows between the fragment and the fetched page is gone. + const raw = await server.callTool('lens_list_pairs', { limit: config.mcp.maxPageLimit }) + expect(raw.structuredContent).toBeUndefined() + const body = JSON.parse(raw.content[0].text) + expect(body.truncated).toBe(true) + expect(body.nextCursor).toBe(null) + expect(body.pairs.length).toBeLessThan(config.mcp.maxPageLimit) + expect(body.pairs.length).toBeGreaterThan(0) + expect(body.notice).toMatch(/smaller limit/) + // Every row the agent does receive still carries a full issuer. + expect(body.pairs[0].assetB.issuer).toHaveLength(56) + }) + + it('scopes a named network to that network and to that network\'s pairs', async () => { + mockGetActivePairs.mockReturnValue(makePairs(1)) + mockLatestPrices() + const server = buildServer() + const app = await buildRestApp() + + await app.inject({ method: 'GET', url: '/pairs?network=mainnet' }) + expect(mockQuery.mock.calls[0][1][0]).toBe('mainnet') + + const mainnet = await callToolJson(server, 'lens_list_pairs', { network: 'mainnet' }) + expect(mockQuery.mock.calls.at(-1)![1][0]).toBe('mainnet') + + // The registry belongs to whichever network this process serves, so asking + // for the other one gets that network's declared pairs — a testnet listing + // must never be read as a mainnet one. + const keys = mainnet.pairs.map((p: { pairKey: string }) => p.pairKey) + expect(keys).toEqual(getNetworkConfig('mainnet').pairs.map(p => p.pairKey)) + expect(keys).not.toContain(makePairs(1)[0].pairKey) + await app.close() + }) +}) + +// ── lens_get_pools ──────────────────────────────────────────────────────────── + +describe('lens_get_pools vs GET /pools', () => { + // `numeric` columns arrive from the driver as strings, which is what lets + // `?decimalFormat=string` return the stored value exactly. + const POOLS = [ + { pool_id: 'pool-a', asset_a: 'XLM', asset_b: 'USDC', reserve_a: '10.5', reserve_b: '20', spot_price: '2', fee_bp: 30, timestamp: OLD }, + { pool_id: 'pool-b', asset_a: 'XLM', asset_b: 'USDC', reserve_a: '11', reserve_b: '21', spot_price: '2', fee_bp: 30, timestamp: OLD }, + { pool_id: 'pool-c', asset_a: 'XLM', asset_b: 'USDC', reserve_a: '12', reserve_b: '22', spot_price: '2', fee_bp: 30, timestamp: OLD }, + ] + + it('returns the same pools as the route, each asset with its full issuer', async () => { + mockGetActivePairs.mockReturnValue(WATCHED) + // The DB answers the keyset pages: everything for the REST route and page 1, + // then only what follows pool-b for page 2. + mockQuery + .mockResolvedValueOnce({ rows: POOLS }) + .mockResolvedValueOnce({ rows: POOLS }) + .mockResolvedValueOnce({ rows: POOLS.slice(2) }) + const server = buildServer() + const app = await buildRestApp() + + const rest = (await app.inject({ method: 'GET', url: '/pools' })).json() + expect(rest.pools).toHaveLength(3) + expect(rest.pools[0].reserve_a).toBe(10.5) + + const page1 = await callToolJson(server, 'lens_get_pools', { limit: 2 }) + expect(page1.pools).toHaveLength(2) + // Every REST field is present and equal, plus the resolved full asset ids. + expect(page1.pools[0]).toEqual({ + ...rest.pools[0], + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: USDC_ISSUER }, + }) + expect(page1.hasMore).toBe(true) + expect(typeof page1.nextCursor).toBe('string') + + // The cursor is a keyset on pool_id: the next page starts after pool-b. + const page2 = await callToolJson(server, 'lens_get_pools', { limit: 2, cursor: page1.nextCursor }) + const [sql, params] = mockQuery.mock.calls.at(-1)! + expect(sql).toMatch(/pool_id > \$2/) + expect(params).toEqual([activeNetwork, 'pool-b', 3]) + expect(page2.pools).toHaveLength(1) + expect(page2.pools[0].pool_id).toBe('pool-c') + expect(page2.hasMore).toBe(false) + await app.close() + }) + + it('falls back to a null issuer for a pool whose pair is no longer watched', async () => { + mockGetActivePairs.mockReturnValue(WATCHED) + mockQuery.mockResolvedValue({ + rows: [{ pool_id: 'pool-x', asset_a: 'FOO', asset_b: 'BAR', reserve_a: '1', reserve_b: '2', spot_price: '2', fee_bp: 30, timestamp: OLD }], + }) + const server = buildServer() + + // A pool whose codes match no watched pair keeps null issuers rather than + // borrowing the only one on hand: a wrong issuer is a different asset. + const page = await callToolJson(server, 'lens_get_pools', {}) + expect(page.pools[0].assetA).toEqual({ code: 'FOO', issuer: null }) + expect(page.pools[0].assetB).toEqual({ code: 'BAR', issuer: null }) + }) + + it('clamps an oversized pool page to the server ceiling', async () => { + mockGetActivePairs.mockReturnValue(WATCHED) + mockQuery.mockResolvedValue({ rows: POOLS }) + const server = buildServer() + + const capped = await callToolJson(server, 'lens_get_pools', { limit: 500 }) + // limit 500 is clamped to the ceiling, so the query asks for one more than that. + expect(capped.limit).toBe(config.mcp.maxPageLimit) + expect(mockQuery.mock.calls.at(-1)![1].at(-1)).toBe(config.mcp.maxPageLimit + 1) + }) +}) + +// ── lens_screen_pairs ───────────────────────────────────────────────────────── + +describe('lens_screen_pairs vs GET /screener', () => { + const ROWS = [ + { pair_key: `USDC:${USDC_ISSUER}/XLM`, volume: 30, price: 0.12, change_24h: 1.5, liquidity: 50000 }, + { pair_key: 'A000:GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA/XLM', volume: 20, price: 0.5, change_24h: -2, liquidity: 40000 }, + { pair_key: 'B001:GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA/XLM', volume: 10, price: 0.8, change_24h: 0, liquidity: 30000 }, + ] + + it('returns exactly the route body for the same filters, sort and page', async () => { + mockQuery.mockResolvedValue({ rows: ROWS }) + const server = buildServer() + const app = await buildRestApp() + + const rest = await app.inject({ + method: 'GET', + url: '/screener?volume=10&sortBy=liquidity&order=asc&limit=2', + }) + expect(rest.statusCode).toBe(200) + const restBody = rest.json() + + const toolBody = await callToolJson(server, 'lens_screen_pairs', { + volume: 10, + sortBy: 'liquidity', + order: 'asc', + limit: 2, + }) + // The route renders the service page as-is; the tool adds the page size it + // actually ran, because a clamped ask is not something an agent should have + // to infer from a short page. Everything else is byte-identical. + expect(toolBody).toEqual({ ...restBody, limit: 2 }) + + // Not just equal output: the identical SQL and bound parameters, because the + // tool runs the same service the route and GraphQL do. + expect(mockQuery.mock.calls[1]).toEqual(mockQuery.mock.calls[0]) + + // pair_key carries the full issuer, never a shortened one. + expect(toolBody.data[0].pair_key).toContain(USDC_ISSUER) + await app.close() + }) + + it('pages with the same cursor as the route', async () => { + mockQuery + .mockResolvedValueOnce({ rows: ROWS }) + .mockResolvedValueOnce({ rows: ROWS.slice(2) }) + const server = buildServer() + + const page1 = await callToolJson(server, 'lens_screen_pairs', { limit: 2 }) + expect(page1.hasMore).toBe(true) + expect(page1.data).toHaveLength(2) + + const page2 = await callToolJson(server, 'lens_screen_pairs', { limit: 2, cursor: page1.nextCursor }) + expect(page2.data).toHaveLength(1) + expect(page2.hasMore).toBe(false) + expect(page2.nextCursor).toBe(null) + }) + + it('refuses market_cap with the message REST uses', async () => { + const server = buildServer() + const response = await server.callTool('lens_screen_pairs', { market_cap: '1000' }) + expect(response.isError).toBe(true) + expect(JSON.parse(response.content[0].text).reason).toBe(MARKET_CAP_UNSUPPORTED) + }) +}) + +// ── Error mapping ───────────────────────────────────────────────────────────── + +describe('lens tool errors', () => { + it('rejects an unknown network with ERR_BAD_REQUEST', async () => { + const server = buildServer() + const response = await server.callTool('lens_list_pairs', { network: 'devnet' }) + expect(response.isError).toBe(true) + expect(JSON.parse(response.content[0].text).code).toBe('ERR_BAD_REQUEST') + }) + + it('rejects an argument it does not take rather than ignoring it', async () => { + const server = buildServer() + const response = await server.callTool('lens_list_pairs', { limitd: 5 }) + expect(response.isError).toBe(true) + const payload = JSON.parse(response.content[0].text) + expect(payload.code).toBe('ERR_BAD_REQUEST') + expect(payload.reason).toMatch(/limitd/) + }) + + it('rejects a malformed cursor with ERR_BAD_REQUEST', async () => { + mockGetActivePairs.mockReturnValue(makePairs(1)) + mockLatestPrices() + const server = buildServer() + + for (const name of ['lens_list_pairs', 'lens_get_pools', 'lens_screen_pairs']) { + const response = await server.callTool(name, { cursor: 'not-base64-json' }) + expect(response.isError, `${name} accepted a malformed cursor`).toBe(true) + expect(JSON.parse(response.content[0].text).code).toBe('ERR_BAD_REQUEST') + } + }) + + it('rejects a non-numeric filter with the screener validation message', async () => { + const server = buildServer() + const response = await server.callTool('lens_screen_pairs', { volume: 'lots' }) + expect(response.isError).toBe(true) + const payload = JSON.parse(response.content[0].text) + expect(payload.code).toBe('ERR_BAD_REQUEST') + expect(payload.reason).toMatch(/volume must be a valid number/) + }) + + it('maps a database failure to ERR_SERVICE_UNREACHABLE without leaking it', async () => { + mockGetActivePairs.mockReturnValue(makePairs(1)) + mockQuery.mockRejectedValue(new Error('connection refused postgres://user:secret@host')) + const spy = vi.spyOn(console, 'error').mockImplementation(() => {}) + const server = buildServer() + + const response = await server.callTool('lens_list_pairs', {}) + expect(response.isError).toBe(true) + const payload = JSON.parse(response.content[0].text) + expect(payload.code).toBe('ERR_SERVICE_UNREACHABLE') + // A tool reply lands in an agent's transcript, so the driver error — which + // carries the connection string — belongs in the log and nowhere else. + expect(response.content[0].text).not.toMatch(/secret/) + expect(spy).toHaveBeenCalled() + spy.mockRestore() + }) +})