[gandhipratik203]

Summary

• Add an optional server_id to ToolPreviewRequest so callers can preview a tool through a virtual-server scope.
• Verify that the caller can access the requested virtual server before resolving the tool.
• Pass server_id to the existing preview service so only tools attached to that server are considered.
• Preserve the existing global preview behavior when server_id is omitted or null.

Note: This PR extends the existing tool preview endpoint. It does not add the endpoint, preview execution logic, or preview_safe hook filtering.

Explanatory Diagrams

Tool Preview Before vs After This PR

Before
======

Preview request
  tool name + arguments
          |
          v
Search all visible tools globally
          |
          v
May resolve a different tool than a
virtual-server-scoped live invocation


After
=====

Preview request
  tool name + arguments + server_id
                    |
                    v
Check access to the virtual server
                    |
                    v
Search only tools attached to that server
                    |
                    v
Resolve the same scoped tool as live invocation

Compatibility

• Requests without server_id, or with server_id: null, continue to preview globally.
• The endpoint requires the existing tools.preview permission.
• A server-scoped token currently receives 403 from token-scoping middleware on this REST route, including when the request names the token's own server. Changing that middleware behavior is outside this PR.

Manual Verification

Verified with frontend main and backend PR #7249.

The run uses the existing SQLite mcp.db and real backend APIs. Browser Preview and Live invocation are performed manually through the BFF.

The test uses a temporary Python environment with the backend's lockfile dependencies. The steps support either that environment or an equivalent .venv.

Setup and test steps

Before starting

Use frontend main, the backend PR branch, and dependencies installed from their lockfiles. Use your existing admin account.

Commands default to the backend's .venv. For another environment, export PR7249_PYTHON with its Python path in Terminals A and D.

Replace the paths below. Keep ports 8000, 5173, 3000, and 9010 free. Run each service in its own terminal; keep Terminal D open for the helper commands.

Use a fresh state directory for each run. Keep PR7249_STATE_DIR set in Terminal D for every helper command.

1. Start backend (Terminal A)

cd /path/to/mcp-context-forge
export PR7249_PYTHON="${PR7249_PYTHON:-$PWD/.venv/bin/python}"
MCPGATEWAY_TOOL_PREVIEW_ENABLED=true SSRF_ALLOW_LOCALHOST=true "$PR7249_PYTHON" -m uvicorn mcpgateway.main:app --host 127.0.0.1 --port 8000

This uses your existing .env and mcp.db. If startup rejects placeholder secrets, first run "$PR7249_PYTHON" -m mcpgateway.scripts.init_secrets --patch-env .env.

2. Start frontend (Terminal B)

cd /path/to/contextforge-web-ui
HUSKY=0 npm ci
HUSKY=0 npm ci --prefix server
npm run generate
npm run dev -- --host 127.0.0.1 --port 5173 --strictPort

3. Start BFF (Terminal C)

cd /path/to/contextforge-web-ui
CONTEXTFORGE_URL=http://127.0.0.1:8000 COOKIE_SECURE=false HOST=127.0.0.1 PORT=3000 REDIS_URL=memory:// SSO_ENABLED=false PUBLIC_ORIGIN=http://localhost:3000 VITE_DEV_SERVER_URL=http://127.0.0.1:5173 npm run dev --prefix server

Keep Terminals A-C running. Open http://localhost:3000 and sign in.

4. Create test fixtures (Terminal D)

Put the script from the next collapsible in the backend repo root as pr7249-manual-test.py. Keep it untracked.

cd /path/to/mcp-context-forge
export PR7249_PYTHON="${PR7249_PYTHON:-$PWD/.venv/bin/python}"
export PR7249_STATE_DIR="$(mktemp -d "${TMPDIR:-/tmp}/pr7249-manual.XXXXXX")"
"$PR7249_PYTHON" pr7249-manual-test.py --state-dir "$PR7249_STATE_DIR" setup

Enter your admin password. For another account, append --email your-admin@example.com to setup.

Expect one MCP echo source and two virtual servers: pr7249-attached-... has echo; pr7249-empty-... does not. Upstream executions starts at 0.

5. Check API preview

"$PR7249_PYTHON" pr7249-manual-test.py --state-dir "$PR7249_STATE_DIR" check

Expect attached preview 200, unattached preview 404, and execution count 0 -> 0.

6. Test browser Preview

Open Virtual Servers -> pr7249-attached-... -> View details -> Components -> Echo actions -> Test.

Enter hello, leave Live invocation off, and click Preview. Expect Preview 200, validated: true, and the expected source. Check that the request snippet includes this server's server_id.

"$PR7249_PYTHON" pr7249-manual-test.py --state-dir "$PR7249_STATE_DIR" status

Expect Upstream executions: 0.

7. Test Live invocation

Turn Live invocation on. Keep message=hello and click Invoke tool once. Expect Live invoke 200, the attached server in Requested through, and hello from the same source as Preview.

"$PR7249_PYTHON" pr7249-manual-test.py --state-dir "$PR7249_STATE_DIR" status

Expect Upstream executions: 1 and Message: hello.

8. Cleanup

Keep the backend running:

"$PR7249_PYTHON" pr7249-manual-test.py --state-dir "$PR7249_STATE_DIR" cleanup

This removes this run's fixtures and stops its echo server. State and logs remain. Use a fresh state directory next time. Stop Terminals A-C with Ctrl-C when finished.

Supporting local helper script

Save this as pr7249-manual-test.py in the backend root. Keep it untracked.

The helper creates fixtures and checks Preview through backend APIs. Perform browser Preview and Live invocation manually.

Cleanup checks whether gateway deletion already removed the tool before deleting it separately.

#!/usr/bin/env python3
"""Local real-backend fixtures for PR #7249; keep this helper untracked.

Run with the backend's installed Python environment:
    pr7249-manual-test.py setup
    pr7249-manual-test.py status
    pr7249-manual-test.py check
    pr7249-manual-test.py cleanup

Start the backend separately. Setup asks for your existing login password.
Use a fresh --state-dir for each run. The echo service defaults to port 9010.
Perform browser Preview and Live invocation yourself; this helper does not run them.
The upstream runs in the background until cleanup. State and logs live in the
OS temporary directory. No database writes bypass the backend HTTP API.
"""

from __future__ import annotations

import argparse
import getpass
import json
import os
from pathlib import Path
import socket
import subprocess
import sys
import tempfile
import time
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlsplit
from urllib.request import Request, urlopen
import uuid


SCRIPT = Path(__file__).resolve()
DEFAULT_STATE = Path(tempfile.gettempdir()) / 'pr7249-manual-test'


def save(directory, state):
    directory.mkdir(mode=0o700, parents=True, exist_ok=True)
    pending = directory / 'state.pending'
    descriptor = os.open(pending, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(descriptor, 'w') as output:
        json.dump(state, output, indent=2)
        output.write('\n')
    pending.replace(directory / 'state.json')


def load(directory):
    path = directory / 'state.json'
    if not path.exists():
        raise RuntimeError('No test state found. Run setup first.')
    state = json.loads(path.read_text())
    if state.get('script') != str(SCRIPT):
        raise RuntimeError('State belongs to a different helper; use its original script.')
    return state


def http(method, url, *, body=None, token=None, headers=None, expected=(200,)):
    request_headers = dict(headers or {})
    if token:
        request_headers['Authorization'] = f'Bearer {token}'
    data = None
    if body is not None:
        request_headers['Content-Type'] = 'application/json'
        data = json.dumps(body).encode()
    try:
        response = urlopen(Request(url, data=data, headers=request_headers, method=method), timeout=30)
    except HTTPError as error:
        response = error
    with response:
        raw = response.read()
        payload = json.loads(raw) if raw else None
        status = response.status
    if status not in expected:
        raise RuntimeError(f'{method} {urlsplit(url).path}: HTTP {status}; expected {expected}')
    return status, payload


def login(directory, state):
    password = getpass.getpass(f'Existing backend password for {state["email"]}: ')
    _, auth = http('POST', state['backend'] + '/auth/email/login', body={'email': state['email'], 'password': password})
    state['token'] = auth['access_token']
    save(directory, state)


def api(directory, state, method, path, *, body=None, expected=(200,)):
    status, payload = http(method, state['backend'] + path, body=body, token=state.get('token'), expected=(*expected, 401))
    if status == 401:
        login(directory, state)
        return http(method, state['backend'] + path, body=body, token=state['token'], expected=expected)
    return status, payload


def calls(directory):
    path = directory / 'calls.jsonl'
    return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []


def upstream_stats(state):
    _, stats = http('GET', state['upstream'] + '/stats')
    if stats.get('run_id') != state['run_id']:
        raise RuntimeError('Upstream belongs to another run; refusing to use it.')
    return stats


def serve(directory):
    from mcp.server.mcpserver import MCPServer
    from mcp.types import ToolAnnotations
    from starlette.responses import JSONResponse
    import uvicorn

    state = load(directory)
    mcp = MCPServer(state['gateway_name'])

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False))
    async def echo(message: str) -> dict:
        """Return a message from the real PR7249 test upstream."""
        with (directory / 'calls.jsonl').open('a') as output:
            output.write(json.dumps({'message': message, 'time': time.time()}) + '\n')
        return {'message': message, 'source': state['gateway_name']}

    application = mcp.streamable_http_app(host='127.0.0.1', stateless_http=True, json_response=True)
    server = None

    async def app(scope, receive, send):
        if scope['type'] == 'http' and scope['path'] == '/stats':
            history = calls(directory)
            response = JSONResponse({'run_id': state['run_id'], 'executions': len(history), 'calls': history})
            await response(scope, receive, send)
            return
        if scope['type'] == 'http' and scope['path'] == '/shutdown':
            headers = dict(scope.get('headers', []))
            allowed = scope['method'] == 'POST' and headers.get(b'x-pr7249-run') == state['run_id'].encode()
            await JSONResponse({'stopping': allowed}, status_code=200 if allowed else 403)(scope, receive, send)
            if allowed:
                server.should_exit = True
            return
        await application(scope, receive, send)

    server = uvicorn.Server(uvicorn.Config(app, host='127.0.0.1', port=state['port'], log_level='warning'))
    server.run()


def setup(args, directory):
    if (directory / 'state.json').exists():
        raise RuntimeError('Test state already exists. Use status or cleanup; choose a new --state-dir for another run.')
    backend = args.backend.rstrip('/')
    if urlsplit(backend).hostname not in {'127.0.0.1', 'localhost', '::1'}:
        raise RuntimeError('This helper only prepares a local backend.')
    http('GET', backend + '/health')
    with socket.socket() as listener:
        listener.bind(('127.0.0.1', args.port))
    run_id = uuid.uuid4().hex[:10]
    state = {
        'script': str(SCRIPT), 'run_id': run_id, 'backend': backend,
        'email': args.email, 'port': args.port,
        'upstream': f'http://127.0.0.1:{args.port}',
        'gateway_name': f'pr7249-echo-{run_id}',
        'servers': [], 'tool_ids': [], 'gateway_id': None,
    }
    save(directory, state)
    login(directory, state)
    with (directory / 'upstream.log').open('a') as log:
        child = subprocess.Popen(
            [sys.executable, str(SCRIPT), '--state-dir', str(directory), 'serve'],
            stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT,
            start_new_session=True, cwd=SCRIPT.parent,
        )
    state['upstream_pid'] = child.pid
    save(directory, state)
    deadline = time.monotonic() + 15
    while True:
        if child.poll() is not None:
            raise RuntimeError(f'Upstream exited; inspect {directory / "upstream.log"}.')
        try:
            upstream_stats(state)
            break
        except (URLError, TimeoutError):
            if time.monotonic() >= deadline:
                raise RuntimeError('Upstream did not start within 15 seconds.') from None
            time.sleep(0.1)
    _, gateway = api(directory, state, 'POST', '/v1/gateways', body={
        'name': state['gateway_name'], 'url': state['upstream'] + '/mcp/',
        'transport': 'STREAMABLEHTTP', 'visibility': 'public',
        'description': 'Disposable real MCP upstream for scoped preview verification.',
    }, expected=(200, 201, 202))
    state['gateway_id'] = gateway['id']
    save(directory, state)
    deadline = time.monotonic() + 45
    while True:
        _, payload = api(directory, state, 'GET', f'/v1/tools?gateway_id={quote(gateway["id"], safe="")}&limit=100')
        tools = payload if isinstance(payload, list) else payload.get('tools', [])
        if tools:
            break
        if time.monotonic() >= deadline:
            raise RuntimeError('Gateway discovery did not return echo; inspect backend logs and run cleanup.')
        time.sleep(0.5)
    if len(tools) != 1 or tools[0].get('originalName', tools[0].get('original_name')) != 'echo':
        raise RuntimeError('Unexpected tools discovered; run cleanup before retrying.')
    state['tool_ids'] = [tools[0]['id']]
    state['tool_name'] = tools[0]['name']
    save(directory, state)
    for label, tool_ids in [('attached', state['tool_ids']), ('empty', [])]:
        name = f'pr7249-{label}-{run_id}'
        _, server = api(directory, state, 'POST', '/v1/servers', body={
            'server': {'name': name, 'associated_tools': tool_ids, 'description': 'Disposable scoped-preview test fixture.'},
            'visibility': 'public',
        }, expected=(200, 201))
        state['servers'].append({'id': server['id'], 'name': name, 'kind': label})
        save(directory, state)
    status(directory, state)
    print('Open http://localhost:3000 -> Virtual Servers -> attached server -> Components -> echo -> Test.')
    print('Preview must leave the execution count unchanged; Live invocation should echo your message.')


def status(directory, state):
    print('State:', directory)
    print('Backend:', state['backend'])
    print('MCP source:', state['gateway_name'])
    print('Qualified tool name:', state.get('tool_name', 'not discovered yet'))
    for server in state['servers']:
        print(f'Virtual server ({server["kind"]}): {server["name"]} [{server["id"]}]')
    try:
        stats = upstream_stats(state)
        print('Upstream executions:', stats['executions'])
        for call in stats['calls']:
            print('  Message:', call['message'])
    except (OSError, RuntimeError):
        print('Upstream is not reachable or does not belong to this run.')


def check(directory, state):
    servers = {server['kind']: server for server in state['servers']}
    if not state.get('tool_name') or not {'attached', 'empty'} <= servers.keys():
        raise RuntimeError('Setup is incomplete; inspect status and run cleanup before retrying.')
    before = upstream_stats(state)['executions']
    path = '/v1/tools/preview/' + quote(state['tool_name'], safe='')
    code, preview = api(directory, state, 'POST', path, body={
        'server_id': servers['attached']['id'], 'arguments': {'message': 'preview-only'},
    })
    target = preview.get('target', {})
    if preview.get('validated') is not True or target.get('gatewayName', target.get('gateway_name')) != state['gateway_name']:
        raise RuntimeError('Attached preview did not validate and resolve to the expected upstream.')
    print('Attached scoped preview:', code, '- expected MCP source confirmed')
    code, _ = api(directory, state, 'POST', path, body={
        'server_id': servers['empty']['id'], 'arguments': {'message': 'must-not-run'},
    }, expected=(404,))
    print('Unattached scoped preview:', code)
    after = upstream_stats(state)['executions']
    if before != after:
        raise RuntimeError(f'Preview executed the upstream: {before} -> {after}')
    print(f'Preview execution count unchanged: {before} -> {after}')
    print('These are real API checks. Browser Preview and Live invocation are separate manual steps.')


def cleanup(directory, state):
    for server in list(state['servers']):
        api(directory, state, 'DELETE', '/v1/servers/' + quote(server['id'], safe=''), expected=(200, 204, 404))
        state['servers'].remove(server)
        save(directory, state)
    if state.get('gateway_id'):
        gateway_id = state['gateway_id']
        api(directory, state, 'DELETE', '/v1/gateways/' + quote(gateway_id, safe=''), expected=(200, 202, 204, 404))
        deadline = time.monotonic() + 45
        while True:
            code, _ = api(directory, state, 'GET', '/v1/gateways/' + quote(gateway_id, safe=''), expected=(200, 404))
            if code == 404:
                break
            if time.monotonic() >= deadline:
                raise RuntimeError('Gateway deletion is still pending; rerun cleanup.')
            time.sleep(0.5)
        state['gateway_id'] = None
        save(directory, state)
    for tool_id in list(state['tool_ids']):
        path = '/v1/tools/' + quote(tool_id, safe='')
        code, _ = api(directory, state, 'GET', path, expected=(200, 404))
        if code != 404:
            api(directory, state, 'DELETE', path, expected=(200, 204, 404))
        state['tool_ids'].remove(tool_id)
        save(directory, state)
    try:
        upstream_stats(state)
    except URLError:
        pass
    else:
        http('POST', state['upstream'] + '/shutdown', headers={'X-PR7249-Run': state['run_id']})
    state.pop('token', None)
    state['cleaned'] = True
    save(directory, state)
    print('Created fixtures removed; upstream stopped. State and logs retained at', directory)
    print('Use a new --state-dir for another run.')


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--state-dir', type=Path, default=DEFAULT_STATE)
    commands = parser.add_subparsers(dest='command', required=True)
    prepare = commands.add_parser('setup', help='Start the real upstream and create disposable API fixtures.')
    prepare.add_argument('--backend', default='http://127.0.0.1:8000')
    prepare.add_argument('--email', default='admin@example.com')
    prepare.add_argument('--port', type=int, default=9010)
    commands.add_parser('status', help='Print fixture names and real upstream execution count.')
    commands.add_parser('check', help='Check attached/unattached preview without executing echo.')
    commands.add_parser('cleanup', help='Delete only recorded fixtures and stop their upstream.')
    commands.add_parser('serve', help='Internal upstream worker started by setup.')
    args = parser.parse_args()
    directory = args.state_dir.expanduser().resolve()
    try:
        if args.command == 'setup':
            setup(args, directory)
        elif args.command == 'serve':
            serve(directory)
        else:
            {'status': status, 'check': check, 'cleanup': cleanup}[args.command](directory, load(directory))
    except (OSError, RuntimeError, ValueError, KeyError) as error:
        print(f'Error: {error}', file=sys.stderr)
        print(f'State directory: {directory}', file=sys.stderr)
        return 1
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

Observed Results

The user completed the browser steps and provided screenshots. The helper confirmed execution counts before and after Live invocation.

Check	Observed result
Attached scoped preview through the API	200; expected MCP source confirmed
Same tool through the empty virtual server	404
API preview execution count	Unchanged: 0 -> 0
Manual browser Preview with message=hello	Preview 200; validated: true; expected source and server ID in the request snippet
Execution count after browser Preview	0
Manual browser Live invocation with message=hello	Live invoke 200; returns hello from the same source through the attached server
Execution count after one browser Live invocation	1; recorded message hello
Cleanup	Helper completes; saved test token removed; echo service on port 9010 stops

Recorded fixture names:

MCP source:       pr7249-echo-082968dda2
Qualified tool:   pr7249-echo-082968dda2-echo
Attached server:  pr7249-attached-082968dda2
Empty server:     pr7249-empty-082968dda2

The backend, frontend, and BFF remain running after fixture cleanup. Local test logs are retained.

This manual run uses one upstream tool. It does not repeat duplicate-name, access-denial, or server-scoped-token tests.

Closes #6743

[msureshkumar88]

Thank you for the PR. The code change is small, correct, and backend-only. It adds the missing server_id wiring that #6743 describes, and it needs no Alembic migration. The integration tests pass locally (31 passed with --with-integration).

No finding meets a blocking criterion. I request changes on the PR text, the docs, and the test coverage of the security-sensitive branches.

Required before merge

functionally-impacting | PR title and body | The title and body say this PR implements the endpoint, the ToolService preview logic, and preview_safe hook filtering. Those exist already (#5629). The diff does not touch tool_service.py. The body says the route requires tools.execute. It requires tools.preview. | Retitle to feat: accept server_id on tool preview (#6743). Describe the three real changes: the schema field, the route wiring, and the server visibility check.

functionally-impacting | PR body, "Closes" section | ## Closes followed by #6743 on the next line does not link the issue. GitHub reports no closing reference. | Write Closes #6743 on one line.

functionally-impacting | docs/docs/manage/api-usage.md (Preview a Tool Call) | The section does not document server_id, the new status codes, or an example. CHANGELOG.md has no entry. | Document the field and add an example. List 403 (token scope), 404 (unknown or hidden server), 404 (tool not attached), and 400 (ambiguous upstream name). Add a CHANGELOG line. State that an older gateway ignores server_id and previews globally.

functionally-impacting | tests/integration/test_tool_preview.py:363-411 | The server visibility deny paths mock ensure_server_access. No real-database test covers a nonexistent server, another user's private server, or a wrong-team server. These are the security-sensitive branches. | Add three real-database tests that expect the same generic 404. Follow TestToolPreviewWrongTeamIntegration.

functionally-impacting | tests/integration/test_tool_preview.py:336-387 | The 403 test patches extract_token_scopes_from_request. TokenScopingMiddleware._check_server_restriction default-denies /tools/preview for every server-scoped token, so a real server-scoped token returns 403 in the middleware and never reaches the new handler branch. The test describes behavior that real traffic cannot produce. | Add one middleware-level test that pins the 403 for a server-scoped token. State the limit in the docs. A server-scoped token cannot preview its own server today.

Should fix

suggestion | mcpgateway/main.py:5901-5905 | extract_token_scopes_from_request re-verifies the JWT and returns None on any exception. validate_server_access(None, ...) then returns True, so a failure grants access (CWE-755). /rpc reads the payload that auth already cached in request.state._jwt_verified_payload. This is the only route-level caller of the helper. | Read the cached payload. Move the "scope check, then ensure_server_access" sequence into one helper that /rpc and preview share. Deny when a token is present and its scopes cannot be read.

suggestion | mcpgateway/schemas.py:2175,2190-2207 | server_id has no length cap. A 100 000-character value passes validation, reaches a SQL bind parameter, and returns in the 403 and 404 detail strings (CWE-770). Server.id is String(36). The validator strips whitespace, but /rpc does not. The isinstance(v, str) test is dead code. | Use Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=36)] and delete the validator.

suggestion | tests/integration/test_tool_preview.py:733-747 | The duplicate-name test asserts 400, then two 200 responses. It does not prove that each qualified name resolved to a different gateway. | Assert target.gatewayName equals preview-source-one and preview-source-two.

suggestion | tests/unit/mcpgateway/test_main.py:1515-1552 | No handler test covers server_id. Integration tests skip unless --with-integration is set, so make coverage diff-cover may miss the new main.py lines. | Add two handler-level cases for the 403 and 404 branches.

Optional

• tests/unit/mcpgateway/test_tool_preview_schema.py: 168 lines for one field. Several tests restate Pydantic behavior. Collapse to a short parametrized test.
• tests/live_gateway/: AGENTS.md asks for a black-box test. Add one in this PR or a follow-up.
• Squash the two docs: follow-up commits. Restore the (#5629) reference removed from the preview_tool docstring.
• Replace the repeated inline db_mod.Tool(...) blocks with a factory helper.

Compatibility

The change is additive. Clients that already send server_id (ignored today because of extra="ignore") now get a scoped result, 403, 404, or 422. A new client on an older gateway gets a silent global preview. This is the divergence #6743 targets. Document the minimum gateway version.

Notes

• .secrets.baseline change is expected (line shift in schemas.py).
• The server visibility check is necessary. The resolver checks tool access and tool-to-server membership, but not server visibility.

[gandhipratik203]

Updated:

• PR title, description, permission name, and closing reference.
• API docs, examples, errors, and token limitations.
• Real-database denial and middleware tests.
• Verified JWT handling with fail-closed checks.
• server_id length limit and boundary tests.
• Gateway-resolution assertions and handler coverage.

Deferred as outside this PR’s scope:

• Shared /rpc authorization refactoring.
• Test consolidation and factory cleanup.
• A committed live-gateway test; real-backend manual verification is documented.

Changelog updates happen at release time.

All 80 focused tests pass. Broader suites pass separately; the combined run has three plugin-state isolation failures

[msureshkumar88]

The change delivers #6743 and stays additive. The optional server_id, the ensure_server_access preflight, and the real-DB deny-path tests (missing, private, wrong-team) are solid. The docs section is accurate. No Alembic or schema change is needed. Reviewed at 3d2ffb70b.

Two items must change before merge.

functionally-impacting | mcpgateway/main.py:5907 | `payload is None and (token or cached is not None)` returns 403 "Unable to read verified token scopes" for a valid request under proxy auth. | Fail closed only when `cached is not None and payload is None`; drop the `token or` clause.
functionally-impacting | tests/live_gateway/ | No black-box test against a running gateway, and no test compares preview with live `tools/call` for the same name and `server_id` (issue criterion 3). | Add one live-gateway test: two same-name tools on one server, qualified name resolves to the same tool in preview and in `/rpc tools/call`.
suggestion | mcpgateway/main.py:5902 | Third inline copy of the token-scope extraction (`main.py:508`, `11342`). The new copy fails closed; the others do not. | Extract one helper with a strict flag, or record a follow-up issue.
suggestion | tests/integration/test_tool_preview.py:544 | About eight near-identical `db_mod.Tool(...)` blocks. | Generalize `_make_team_tool` into one `_make_tool(...)` factory.
suggestion | tests/unit/mcpgateway/test_tool_preview_schema.py | `test_model_dump`, `test_json_serialization`, and similar tests exercise Pydantic, not this change. | Parametrize the `server_id` validation cases and drop the rest.
minor | mcpgateway/main.py:5871 | The docstring drops `(#5629)`. | Restore it or leave it unchanged.
minor | CHANGELOG.md | No entry. | Add one if the project lists API additions there.
minor | branch | Four commits behind `main`. | Rebase before merge.

Finding 1 detail. The failure needs MCP_CLIENT_AUTH_ENABLED=false, TRUST_PROXY_AUTH=true, and the dangerous-trust flag. The request carries X-Authenticated-User plus Authorization: Bearer <opaque IdP token>. Nothing sets _jwt_verified_payload on that path (rbac.py:277-285), so the same call succeeds without server_id and returns 403 with it. Add a unit test in test_tool_preview_handler.py for this case.

Why the scope block stays: TokenScopingMiddleware registers only when EMAIL_AUTH_ENABLED=true (main.py:3330). With it off, this block is the only server-scope guard. A server-scoped token for the wrong server gets 403, and the token's own server passes.

Two observations for context. The flaky wrong-team-/tools case failed once in my first full run and did not recur in later runs, so I cannot name a cause. The .secrets.baseline change is ignored.

[gandhipratik203]

Addressed both required items:

• Fixed proxy-auth handling and added a regression test.
• Added a live-gateway test confirming Preview/Live parity for duplicate-name tools and zero Preview executions.

77 focused tests and the live test pass. Shared refactoring and optional cleanup remain deferred as outside this PR’s scope.

[msureshkumar88]

Thank you for the follow-up. Both required items from the previous review are resolved. Reviewed at 47d4aff1f.

Resolved

• mcpgateway/main.py:5905: the handler now fails closed only when a cached payload exists but is unreadable. A trusted-proxy request with an opaque bearer token now reaches ensure_server_access. Every JWT path (verify_credentials_cached, external IdP, proxy auth) sets _jwt_verified_payload, so dropping the token or clause does not reopen a fail-open path. test_preview_handler_preserves_trusted_proxy_auth pins this case through the real get_current_user_with_permissions.
• tests/live_gateway/e2e/test_tool_preview_parity.py: the test runs an isolated gateway with two real upstream echo servers. It proves the bare name returns 400, each qualified name resolves to the same source in preview and in /rpc tools/call, and preview runs zero upstream executions. This covers issue criterion 3.

Verification

• Handler, schema, middleware, and integration tests: 73 passed (--with-integration).
• tests/unit/mcpgateway/test_main.py -k preview: 5 passed.
• Live parity test: passed.
• ruff check on changed Python files: clean.

Before merge

minor | .secrets.baseline | The branch conflicts with main in this file only. | Rebase and take the main version, then regenerate with make detect-secrets-scan.

The deferred suggestions (shared scope-extraction helper, _make_tool factory, schema test trim) remain optional.

[vishu-bh]

LGTM 🚀

[vishu-bh]

LGTM