fix: geo_fanout reads the target status via Unlocker format:json (not the gateway 200)

The geo_fanout executor called the Unlocker /request with format:'raw' and then
classified axios response.status, which is always the gateway's 200. A target
403/451/redirect was therefore misclassified as ok, defeating the whole point of
the tool (surfacing geo-gating as a first-class classified result).

Fix: call /request with format:'json'. The response body is the envelope
{status_code, headers, body} where status_code is the TARGET's HTTP status and
headers are the TARGET's response headers. A 3xx is surfaced WITHOUT the Unlocker
following it, so a redirect keeps its Location header. We classify on that real
target status.

- geo_utils.js: add parse_unlocker_json(data) mapping {status_code, headers, body}
  to the {status, headers, body} shape build_geo_entry expects (tolerant of a
  missing/non-object/JSON-string input). build_geo_entry now carries the rendered
  body through and documents exit_ip as best-effort null (the json envelope has no
  gateway x-brd-* headers; we never fabricate an IP).
- server.js: executor uses format:'json' + responseType:'json', parses with
  parse_unlocker_json, passes the target {status, headers} into build_geo_entry,
  and renders the body to markdown via the existing remark/strip pipeline when
  data_format is markdown (raw otherwise). Per-geo country targeting preserved.
- tests: geo-utils fixtures now use the REAL Unlocker envelope shape and drive
  build_geo_entry end-to-end through parse_unlocker_json (403 -> blocked, 302 with
  a cross-host location -> redirected, 200 -> ok, 429 -> rate_limited, thrown
  transport error -> error); parse_unlocker_json gets its own table-driven test.
  The stdio registration test is unchanged.

Based on PR #155 (retry/backoff for #104); retry_utils.js classification is reused.

Author: Yashash4

geo_utils.js

@@ -0,0 +1,198 @@
+'use strict'; /*jslint node:true es9:true*/
+
+// Pure helpers for the geo_fanout tool: validating the list of country exits
+// and turning a set of per-geo results into a single structured report where a
+// blocked / redirected / failed geo is a FIRST-CLASS classified result, not a
+// discarded error. No network or transport dependencies live here so the fanout
+// aggregation logic is unit-testable in isolation.
+
+import {classify_response, OUTCOME} from './retry_utils.js';
+
+// Per-geo status the caller sees in the aggregated report.
+export const GEO_STATUS = {
+    OK: 'ok',               // fetched successfully
+    BLOCKED: 'blocked',     // target refused this geo (403/451)
+    REDIRECTED: 'redirected', // 3xx to a different host/path (geo gating signal)
+    RATE_LIMITED: 'rate_limited',
+    ERROR: 'error',         // transient/fatal failure after retries
+};
+
+// Map the Unlocker's JSON envelope to the shape build_geo_entry expects.
+// When the /request endpoint is called with format:'json', the gateway always
+// answers HTTP 200 and the TARGET's real result lives in the JSON body:
+//   {status_code, headers, body}
+// where status_code is the target's HTTP status and headers are the target's
+// response headers (lowercased keys). The previous format:'raw' path only ever
+// exposed the gateway's 200, so a target 403/451/3xx was misclassified as ok.
+// This helper isolates that mapping so the real envelope shape is unit-testable.
+// Tolerant of a missing / non-object / string-but-JSON input: a value with no
+// recognizable status yields {status:null} and classify_response then treats it
+// as a transport-level failure rather than a silent success.
+export function parse_unlocker_json(data){
+    let obj = data;
+    if (typeof obj=='string')
+    {
+        try { obj = JSON.parse(obj); }
+        catch(e){ obj = null; }
+    }
+    if (!obj || typeof obj!='object')
+        return {status: null, headers: undefined, body: undefined};
+    const status = typeof obj.status_code=='number'
+        && Number.isFinite(obj.status_code) ? obj.status_code : null;
+    const headers = obj.headers && typeof obj.headers=='object'
+        ? obj.headers : undefined;
+    return {status, headers, body: obj.body};
+}
+
+// Validate and normalize a list of 2-letter ISO country codes. Lowercases,
+// trims, dedupes (preserving first-seen order), and rejects malformed entries
+// loudly so a typo never becomes a silently-dropped exit. Returns the clean
+// array; throws Error on any invalid code (callers want fast, explicit failure).
+export function normalize_geos(geos){
+    if (!Array.isArray(geos) || geos.length===0)
+        throw new Error('geos must be a non-empty array of 2-letter codes');
+    const seen = new Set();
+    const out = [];
+    for (const raw of geos)
+    {
+        if (typeof raw!='string')
+            throw new Error(`invalid geo (not a string): ${JSON.stringify(raw)}`);
+        const code = raw.trim().toLowerCase();
+        if (!/^[a-z]{2}$/.test(code))
+            throw new Error(`invalid geo country code: "${raw}" `
+                +`(expected 2 letters, e.g. "us", "de")`);
+        if (seen.has(code))
+            continue;
+        seen.add(code);
+        out.push(code);
+    }
+    return out;
+}
+
+// Detect whether a successful-looking response is actually a geo redirect.
+// A 3xx with a Location header pointing at a different host is the classic
+// "Belgian visitor bounced to a different storefront" signal we must capture.
+function detect_redirect(status, headers, request_url){
+    if (!(status>=300 && status<400))
+        return null;
+    const location = read_header(headers, 'location');
+    if (!location)
+        return {redirected: true, location: null, cross_host: false};
+    let cross_host = false;
+    try {
+        const from = new URL(request_url);
+        const to = new URL(location, request_url);
+        cross_host = from.host!==to.host;
+    } catch(e){
+        cross_host = false;
+    }
+    return {redirected: true, location, cross_host};
+}
+
+function read_header(headers, name){
+    if (!headers || typeof headers!='object')
+        return undefined;
+    const target = name.toLowerCase();
+    for (const key of Object.keys(headers))
+    {
+        if (key.toLowerCase()===target)
+            return headers[key];
+    }
+    return undefined;
+}
+
+// Map one classified outcome to the per-geo report status.
+function outcome_to_geo_status(outcome){
+    switch (outcome)
+    {
+    case OUTCOME.SUCCESS: return GEO_STATUS.OK;
+    case OUTCOME.REDIRECT: return GEO_STATUS.REDIRECTED;
+    case OUTCOME.BLOCKED: return GEO_STATUS.BLOCKED;
+    case OUTCOME.RATE_LIMITED: return GEO_STATUS.RATE_LIMITED;
+    default: return GEO_STATUS.ERROR;
+    }
+}
+
+// Build the per-geo entry for a single settled attempt. `attempt` is the
+// normalized shape produced by the tool's executor:
+//   {geo, url, response:{status, headers, exit_ip?}, body?}  on a completed
+//       request (status/headers are the TARGET's, parsed from the Unlocker
+//       format:'json' envelope, NOT the gateway's 200), or
+//   {geo, url, error:{code?, response?}}                     on a thrown failure.
+// `now_ms` is injectable for deterministic Retry-After math under test.
+export function build_geo_entry(attempt, now_ms = Date.now()){
+    const obj = attempt && typeof attempt=='object' ? attempt : {};
+    const geo = typeof obj.geo=='string' ? obj.geo : 'unknown';
+    const url = typeof obj.url=='string' ? obj.url : null;
+    // Normalize to the shape classify_response understands: a thrown error keeps
+    // its {error} envelope; a completed request passes its {status, headers}.
+    const classify_input = obj.error ? {error: obj.error}
+        : (obj.response || {});
+    const classification = classify_response(classify_input, now_ms);
+    const status = classification.status;
+    const headers = obj.response ? obj.response.headers
+        : (obj.error && obj.error.response
+            ? obj.error.response.headers : undefined);
+    // exit_ip is best-effort: the Unlocker format:'json' envelope carries the
+    // target's headers, not the gateway's x-brd-* headers, so the exit IP is not
+    // observable on this path. We surface whatever the executor explicitly
+    // supplied (none, today) and otherwise null; we never fabricate an IP.
+    const exit_ip = obj.response
+        && typeof obj.response.exit_ip=='string'
+        ? obj.response.exit_ip : null;
+
+    const redirect = detect_redirect(status, headers, url);
+    let geo_status = outcome_to_geo_status(classification.outcome);
+    if (redirect && redirect.redirected)
+        geo_status = GEO_STATUS.REDIRECTED;
+
+    const entry = {
+        geo,
+        url,
+        status: geo_status,
+        http_status: status,
+        exit_ip,
+        outcome: classification.outcome,
+        retry_after_ms: classification.retry_after_ms,
+        reason: classification.reason,
+        redirect: redirect || null,
+    };
+    // The rendered target body (markdown or raw) when the executor captured one.
+    // Kept out of the entry entirely when absent so an error/blocked geo is not
+    // padded with an empty string that reads like real content.
+    if (obj.body!==undefined)
+        entry.body = obj.body;
+    return entry;
+}
+
+// Aggregate all per-geo entries into one report. Every geo appears exactly once
+// regardless of success/failure; nothing is dropped. The summary makes the
+// "this geo was blocked / redirected" fact queryable at a glance.
+export function summarize_fanout(entries){
+    const list = Array.isArray(entries) ? entries : [];
+    const summary = {
+        total: list.length,
+        ok: 0,
+        blocked: 0,
+        redirected: 0,
+        rate_limited: 0,
+        error: 0,
+    };
+    for (const e of list)
+    {
+        switch (e.status)
+        {
+        case GEO_STATUS.OK: summary.ok++; break;
+        case GEO_STATUS.BLOCKED: summary.blocked++; break;
+        case GEO_STATUS.REDIRECTED: summary.redirected++; break;
+        case GEO_STATUS.RATE_LIMITED: summary.rate_limited++; break;
+        default: summary.error++; break;
+        }
+    }
+    return {
+        summary,
+        any_blocked: summary.blocked>0,
+        any_redirected: summary.redirected>0,
+        results: list,
+    };
+}

server.js

@@ -10,6 +10,8 @@ import {parse_google_search_response} from './search_utils.js';
 import {dataset_id_schema, filter_schema, metadata_to_fields, FILTER_OPERATORS}
     from './search_dataset_schema.js';
 import {classify_response, should_retry} from './retry_utils.js';
+import {normalize_geos, parse_unlocker_json, build_geo_entry, summarize_fanout}
+    from './geo_utils.js';
 import {createRequire} from 'node:module';
 import {remark} from 'remark';
 import strip from 'strip-markdown';
@@ -25,7 +27,7 @@ const base_timeout = process.env.BASE_TIMEOUT
 const base_max_retries = Math.max(0,
     Math.min(parseInt(process.env.BASE_MAX_RETRIES || '0', 10) || 0, 3));
 const pro_mode_tools = ['search_engine', 'scrape_as_markdown',
-    'search_engine_batch', 'scrape_batch', 'discover'];
+    'search_engine_batch', 'scrape_batch', 'discover', 'geo_fanout'];
 const tool_groups = process.env.GROUPS ?
     process.env.GROUPS.split(',').map(g=>g.trim().toLowerCase())
         .filter(Boolean) : [];
@@ -74,7 +76,7 @@ if (!api_token)
 
 const sleep = ms=>new Promise(resolve=>setTimeout(resolve, ms));
 
-// Backoff knobs (overridable via env) used by base_request.
+// Backoff knobs (overridable via env) used by base_request and geo_fanout.
 // Issue #104: bursts of MCP calls hit intermittent 502/504 from the gateway and
 // there was no backoff guidance. We now classify each failure and retry only the
 // transient ones with exponential backoff + full jitter, honoring Retry-After.
@@ -429,6 +431,86 @@ addTool({
    }),
 });
 
+addTool({
+    name: 'geo_fanout',
+    description: 'Fetch the SAME url from multiple country exits in parallel and '
+        +'return one structured report. A geo that is blocked (403/451), '
+        +'redirected (3xx to a different host), rate-limited (429) or fails '
+        +'transiently becomes a FIRST-CLASS classified result, not a discarded '
+        +'error. Ideal for detecting geo-gating, regional price/availability '
+        +'differences, and access denial across countries.',
+    annotations: {
+        title: 'Geo Fanout',
+        readOnlyHint: true,
+        openWorldHint: true,
+    },
+    parameters: z.object({
+        url: z.string().url(),
+        countries: z.array(z.string().length(2))
+            .min(1)
+            .max(10)
+            .describe('2-letter ISO country codes to fan the request across '
+                +'(e.g., ["de", "be", "fr"]). Deduped; max 10.'),
+        data_format: z.enum(['raw', 'markdown'])
+            .optional()
+            .default('markdown')
+            .describe('Response body format per geo (default: markdown).'),
+    }),
+    execute: tool_fn('geo_fanout', async({url, countries, data_format}, ctx)=>{
+        const geos = normalize_geos(countries);
+        const now = Date.now();
+        const want_markdown = data_format=='markdown';
+        const attempts = geos.map(geo=>(async()=>{
+            try {
+                // format:'json' makes the Unlocker return a JSON envelope
+                // {status_code, headers, body} where status_code is the TARGET's
+                // HTTP status and headers are the TARGET's response headers. With
+                // the previous format:'raw' we only ever saw the gateway's 200,
+                // so a target 403/451/3xx was misclassified as ok. We classify on
+                // the real target status here. data_format still controls how the
+                // target body is rendered (markdown vs the raw body) inside the
+                // envelope; the 3xx is surfaced in the envelope WITHOUT the
+                // Unlocker following it, so a redirect keeps its Location header.
+                const response = await base_request({
+                    url: 'https://api.brightdata.com/request',
+                    method: 'POST',
+                    data: {
+                        url,
+                        zone: unlocker_zone,
+                        format: 'json',
+                        ...want_markdown ? {data_format: 'markdown'} : {},
+                        country: geo,
+                    },
+                    headers: api_headers(ctx.clientName, 'geo_fanout'),
+                    responseType: 'json',
+                });
+                const parsed = parse_unlocker_json(response.data);
+                let body = parsed.body;
+                if (want_markdown && typeof body=='string' && body)
+                {
+                    body = (await remark()
+                        .use(strip, {keep: ['link', 'linkReference', 'code',
+                            'inlineCode']})
+                        .process(body)).value;
+                }
+                return build_geo_entry({
+                    geo,
+                    url,
+                    body,
+                    response: {
+                        status: parsed.status,
+                        headers: parsed.headers,
+                    },
+                }, now);
+            } catch(e){
+                return build_geo_entry({geo, url, error: e}, now);
+            }
+        })());
+        const entries = await Promise.all(attempts);
+        return JSON.stringify(summarize_fanout(entries), null, 2);
+    }),
+});
+
 addTool({
     name: 'scrape_as_html',
     description: 'Scrape a single webpage URL with advanced options for '

test/geo-fanout-tool.test.js

@@ -0,0 +1,40 @@
+'use strict'; /*jslint node:true es9:true*/
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {fileURLToPath} from 'node:url';
+import {dirname, resolve} from 'node:path';
+import {Client} from '@modelcontextprotocol/sdk/client/index.js';
+import {StdioClientTransport} from '@modelcontextprotocol/sdk/client/stdio.js';
+
+const test_dir = dirname(fileURLToPath(import.meta.url));
+const repo_root = resolve(test_dir, '..');
+
+test('geo_fanout tool is registered and well-formed over stdio', async()=>{
+    const env = {
+        ...process.env,
+        API_TOKEN: 'dummy-token',
+        PRO_MODE: 'true',
+    };
+    const client = new Client(
+        {name: 'geo-fanout-test', version: '0.0.1'},
+        {capabilities: {tools: {}}});
+    const transport = new StdioClientTransport({
+        command: process.execPath,
+        args: ['server.js'],
+        cwd: repo_root,
+        env,
+    });
+    try {
+        await client.connect(transport);
+        const {tools} = await client.listTools();
+        const geo_fanout = tools.find(tool=>tool.name=='geo_fanout');
+        assert.ok(geo_fanout, 'geo_fanout tool is exposed');
+        assert.match(geo_fanout.description, /first-class/i,
+            'description documents first-class classified results');
+        const props = geo_fanout.inputSchema?.properties || {};
+        assert.ok(props.url, 'geo_fanout exposes a url parameter');
+        assert.ok(props.countries, 'geo_fanout exposes a countries parameter');
+    } finally {
+        await client.close();
+    }
+});