# Complete MCP tool tour

Run the maintained complete mcp tool tour against a disposable real engine.

This example uses the [complete Python tour setup](https://beepgrid.com/docs/sdks/python/tour.md) with separate worker credentials. Run from the public engine checkout, with the protected URL/token-file environment and a configured Coordinator provider. The verification runner uses a local deterministic provider fixture. It does not contact real channel providers or execute business tools.

```sh
PYTHONPATH=sdk/python python3 examples/developer-tour/tools.py
```

To run all the tours with generated credentials, a temporary schema and automatic cleanup, use `python3 scripts/documentation-check.py`.



```python
"""Exercise all advertised MCP tools against the same real-engine tour fixture."""
import base64
import json
import os
from pathlib import Path
import runpy
import subprocess
import tempfile
import uuid

state = runpy.run_path('examples/developer-tour/main.py')
owner, builder, billing = (state[name] for name in ['owner', 'builder', 'billing'])
builder_id, billing_id, person_id = (state[name] for name in ['builder_id', 'billing_id', 'person_id'])
run_id = uuid.uuid4().hex
binary = os.getenv('BEEPGRID_DOC_BINARY', '.local/beepgrid')
called = set()

with tempfile.TemporaryDirectory(prefix='beepgrid-mcp-tour-') as scratch:
    builder_file, billing_file = [Path(scratch) / name for name in ['builder-token', 'billing-token']]
    for file, session in [(builder_file, state['builder_session']), (billing_file, state['billing_session'])]:
        file.touch(mode=0o600)
        file.write_text(session['token'])

    def rpc(name, arguments, credential=builder_file):
        env = {**os.environ, 'BEEPGRID_TOKEN': '', 'BEEPGRID_TOKEN_FILE': str(credential), 'BEEPGRID_AGENT_ID': ''}
        request = {'jsonrpc': '2.0', 'id': 1, 'method': 'tools/call', 'params': {'name': name, 'arguments': arguments}}
        process = subprocess.run([binary, 'mcp'], input=json.dumps(request) + '\n', text=True,
                                 capture_output=True, env=env, check=True, timeout=20)
        result = json.loads(process.stdout)['result']
        assert not result['isError'], (name, result)
        called.add(name)
        return json.loads(result['content'][0]['text'])

    # region directory
    assert rpc('list_directory', {'kind': 'agent'})['items']
    assert rpc('list_agents', {})
    assert isinstance(rpc('request_status', {}), list)
    # endregion directory

    # region messages
    sent = rpc('send', {'target': {'kind': 'agent', 'id': billing_id}, 'body': 'Please check the staging ledger.', 'idempotency_key': run_id + '/send'})
    assert rpc('get_conversation', {'id': sent['conversation_id']})['messages']
    assert rpc('start_chat', {'targets': [{'kind': 'agent', 'id': billing_id}], 'body': 'Let us reconcile these orders.', 'idempotency_key': run_id + '/chat'})['conversation_id']
    # endregion messages

    # region membership
    chat = owner.start_chat(targets=[{'kind': 'agent', 'id': builder_id}], body='Review staging together.')
    current = owner.conversation(chat['conversation_id'])
    assert rpc('change_chat_membership', {'conversation_id': chat['conversation_id'], 'operation': 'rename',
        'version': current['version'], 'title': 'Staging review', 'idempotency_key': run_id + '/rename'},
        Path(os.environ['BEEPGRID_TOKEN_FILE']))['version'] > current['version']
    # endregion membership

    # region files
    encoded = base64.b64encode(b'Payment ledger fixture').decode()
    draft = rpc('stage_attachment', {'file_name': 'ledger.txt', 'media_type': 'text/plain', 'base64': encoded})
    with_file = builder.send(target={'kind': 'agent', 'id': billing_id}, body='Review the attached ledger.', attachment_ids=[draft['id']])
    uploaded = rpc('upload_attachment', {'conversation_id': with_file['conversation_id'], 'file_name': 'notes.txt', 'media_type': 'text/plain', 'base64': encoded})
    assert len(rpc('list_attachments', {'conversation_id': with_file['conversation_id']})) == 2
    assert rpc('download_attachment', {'id': uploaded['id']})['base64'] == encoded
    # endregion files

    # region workers
    assert rpc('report_progress', {'progress': {'phase': 'reconciling'}}, billing_file)['ok']
    for delivery in rpc('receive', {}, billing_file):
        assert rpc('acknowledge', {'id': delivery['id'], 'generation': delivery['generation']}, billing_file)['received']
    # endregion workers

    # region decisions
    request = rpc('ask', {'target': {'kind': 'agent', 'id': billing_id}, 'body': 'Which region hosts staging?', 'idempotency_key': run_id + '/ask'})
    current = rpc('get_request', {'id': request['id']}, billing_file)
    answered = rpc('reply', {'id': request['id'], 'version': current['version'], 'outcome': 'answered', 'answer': 'eu-west-1'}, billing_file)
    assert answered['state'] == 'resolved'
    pending = builder.ask(target={'kind': 'agent', 'id': billing_id}, body='I need the latest ledger.')
    assert rpc('set_blocking', {'id': pending['id'], 'blocking': True})['blocking']
    revised = rpc('replace_request', {'id': pending['id'], 'version': pending['version'], 'body': 'I need the ledger for orders 241 and 242.', 'idempotency_key': run_id + '/replace'})
    assert revised['replacement']['state'] == 'open'
    # endregion decisions

    # region execution
    action = {'name': 'fixture', 'executor_id': builder_id, 'resource': 'staging', 'resource_version': 'v2', 'operation_key': run_id + '/fixture', 'arguments': {}}
    approval = owner.ask(target={'kind': 'person', 'id': person_id}, body='Approve this fixture?', kind='approval', action=action)
    owner.respond(approval['id'], 'approved', version=approval['version'])
    execution = rpc('redeem_approval', {'id': approval['id'], 'action': action})
    assert rpc('report_execution', {'id': execution['id'], 'state': 'succeeded', 'result': {'fixture': True}})['state'] == 'succeeded'
    # endregion execution

assert len(called) == 21, called
print('PASS: all 21 MCP tools with actual canonical results')
```