Workspace/Mini projects
Loading progress
All mini projects
MODULE 18 · 5 HOUR BUILD

Versioned collaborator gateway

Create a local gateway that selects an approved compatible collaborator, checks tenant and scope policy, and records task progress and accepted artifacts. The seed is a protocol teaching simulation.

Build evidence Record your actual checks, results, and limitations.

Build it in stages

  1. Define a version-pinned integration manifest separate from discovered metadata.
  2. Implement capability selection and approved-origin checks.
  3. Track remote task IDs and local optimistic state revisions.
  4. Validate artifact schemas and retain declared versus observed capability results.
  5. Add fixtures for incompatible versions, cross-tenant requests, and disconnect-after-acceptance recovery.

Your acceptance criteria

Use these as your project review. Record commands, outputs, and failure cases in your repository.

  • An incompatible version is rejected before dispatch.
  • A tenant mismatch is denied even with a known skill.
  • Completed tasks retain their artifacts after a simulated monitoring disconnect.
  • All simulation-only field names are documented separately from actual A2A schema fields.

A working starting point

The seed runs as supplied. Extend it to satisfy the full brief. It is a teaching starting point, not a finished portfolio submission.

main.py
python
import json

DIRECTORY = {
    'analyst': {'versions': {'1.0'}, 'skills': {'report'},
                'tenants': {'A'}, 'scope': 'report:read'},
}

class Gateway:
    def __init__(self):
        self.tasks = {}

    def submit(self, operation, agent, tenant, version, skill, scopes):
        card = DIRECTORY[agent]
        if version not in card['versions'] or skill not in card['skills']:
            raise ValueError('incompatible collaborator')
        if tenant not in card['tenants'] or card['scope'] not in scopes:
            raise PermissionError('delegation denied')
        if operation not in self.tasks:
            self.tasks[operation] = {'id': 'task-' + operation, 'status': 'working',
                                     'tenant': tenant, 'artifacts': []}
        return self.tasks[operation]['id']

    def complete(self, operation):
        task = self.tasks[operation]
        task['status'] = 'completed'
        task['artifacts'] = [{'type': 'report', 'summary': 'Two checks passed'}]

    def inspect(self, operation, tenant):
        task = self.tasks[operation]
        if task['tenant'] != tenant:
            raise PermissionError('task access denied')
        return json.loads(json.dumps(task))

def main():
    gateway = Gateway()
    task_id = gateway.submit('op7', 'analyst', 'A', '1.0', 'report', {'report:read'})
    gateway.complete('op7')
    print('task:', task_id)
    print('reconnected:', json.dumps(gateway.inspect('op7', 'A'), sort_keys=True))
    print('remote tasks:', len(gateway.tasks))

if __name__ == '__main__':
    main()

Push it further

Replace the local transport with a pinned A2A SDK and run contract tests against a controlled second service, including authentication-required and input-required flows.