#!/usr/bin/env python3 """Read-only Chanceries client and newline-delimited stdio MCP adapter. Python 3.10+; standard library only. No credentials, writes or shell tools. """ from __future__ import annotations import argparse from datetime import date import gzip import json import lzma from pathlib import Path import re import sys import time import unicodedata import urllib.error import urllib.parse import urllib.request ORIGIN_CONFIG = {'canonicalOrigin': 'https://chanceries.org', 'stagingOrigin': 'https://test.chanceries.org', 'legacyProductionOrigins': ['https://chanceries.com', 'https://www.chanceries.com'], 'legacyStagingOrigins': ['https://test.chanceries.com'], 'productionAliases': ['https://www.chanceries.org']} if ORIGIN_CONFIG is None: ORIGIN_CONFIG = json.loads((Path(__file__).resolve().parents[1] / 'data/discovery.json').read_text()) ORIGIN = ORIGIN_CONFIG['canonicalOrigin'] STAGING_ORIGIN = ORIGIN_CONFIG['stagingOrigin'] LEGACY_PRODUCTION_ORIGINS = tuple(ORIGIN_CONFIG.get('legacyProductionOrigins', [])) PRODUCTION_ALIASES = tuple(ORIGIN_CONFIG.get('productionAliases', [])) STAGING_ORIGINS = (STAGING_ORIGIN, *ORIGIN_CONFIG.get('legacyStagingOrigins', [])) ALLOWED_ORIGINS = (ORIGIN, *PRODUCTION_ALIASES, *LEGACY_PRODUCTION_ORIGINS, *STAGING_ORIGINS) ALLOWED_HOSTS = tuple(urllib.parse.urlsplit(origin).netloc for origin in ALLOWED_ORIGINS) MAX_BYTES = 32 * 1024 * 1024 KINDS = ('people', 'appointments', 'countries', 'properties', 'events') CANONICAL_KINDS = ('person', 'appointment', 'polity', 'place', 'event', 'bilateral_relationship', 'topic') RECORD_KINDS = (*KINDS, *CANONICAL_KINDS) def norm(value): return ''.join(c for c in unicodedata.normalize('NFKD', str(value or '')).casefold() if not unicodedata.combining(c)) def canonical_url(kind, record): if record.get('canonicalUrl'): return record['canonicalUrl'] if kind in CANONICAL_KINDS and record.get('canonicalRoute'): return ORIGIN + record['canonicalRoute'] import hashlib rid = record['id'] label = unicodedata.normalize('NFKD', rid.split(':')[-1]).encode('ascii', 'ignore').decode().lower() slug = (re.sub('[^a-z0-9]+', '-', label).strip('-')[:85] or 'record') + '-' + hashlib.sha256(rid.encode()).hexdigest()[:10] if kind == 'people' and record.get('profileUrl'): return ORIGIN + '/diplopedia/people/' + slug + '/' if kind == 'properties': return ORIGIN + '/research/properties/' + slug + '/' if kind == 'appointments': return ORIGIN + '/diplopedia/countries/' + record['slug'] + '/' return ORIGIN + '/' + (record.get('profileUrl') or record.get('file') or '') def year_match(row, year, as_of): """Mirror the site's evidence-aware year filter, without filling gaps.""" if row.get('status') in {'Designate', 'Unverified'}: return '' if row.get('serviceAttestations'): return 'Dated evidence; continuity unconfirmed' if any(int(d['value'][:4]) == year for d in row['serviceAttestations']) else '' start, end = row.get('start'), row.get('end') if row.get('recordedDate') or row.get('dateCoverage') in {'documented-period', 'documented-record', 'recorded-date', 'start-only'}: record = row.get('recordedDate') or start record_end = row.get('recordedEnd') or end or record if record and record_end and int(record['value'][:4]) <= year <= int(record_end['value'][:4]): return 'Dated evidence; continuity unconfirmed' return '' if not start: return '' cutoff = date.fromisoformat(as_of) lo, hi = date(year, 1, 1).toordinal(), min(date(year, 12, 31).toordinal(), cutoff.toordinal()) if lo > hi: return '' if row.get('status') == 'Current': verified = row.get('currentVerifiedAsOf', as_of) if verified: bound = min(cutoff, date.fromisoformat(verified)).toordinal() end = {'min': bound, 'max': bound} else: end = None if not end: return 'Start recorded; departure unknown' if int(start['value'][:4]) == year else '' if start['min'] > hi or end['max'] < lo: return '' return 'Service overlaps this year' if start['max'] <= hi and end['min'] >= lo else 'Possible overlap; dates imprecise' class SameOriginRedirect(urllib.request.HTTPRedirectHandler): def __init__(self, origin): self.origin = urllib.parse.urlsplit(origin) def redirect_request(self, req, fp, code, msg, headers, newurl): target = urllib.parse.urlsplit(newurl) if (target.scheme, target.netloc) != (self.origin.scheme, self.origin.netloc): raise ValueError('Cross-origin data redirect refused') return super().redirect_request(req, fp, code, msg, headers, newurl) class Client: def __init__(self, site_dir=None, base_url=ORIGIN): if base_url not in ALLOWED_ORIGINS: raise ValueError('Use the production origin or the authorized test host') self.origin = ORIGIN if base_url in (*LEGACY_PRODUCTION_ORIGINS, *PRODUCTION_ALIASES) else base_url self.root = Path(site_dir).resolve() if site_dir else None self.opener = urllib.request.build_opener(SameOriginRedirect(self.origin)) self.cache = {} self.loaded_at = 0 self.catalogue = None self.data = {} self.canonical = {'records': [], 'sources': [], 'assertions': [], 'aliases': {}} self.canonical_by_id = {} def read(self, reference): url = urllib.parse.urlsplit(reference) if url.scheme and (url.scheme != 'https' or url.netloc not in ALLOWED_HOSTS): raise ValueError('Untrusted data origin') if url.username or url.password or (url.netloc and not url.scheme) or '%' in url.path: raise ValueError('Invalid data URL') path = url.path.lstrip('/') fixed = {'api/v1/index.json','api/v1/countries.json','api/v1/events.json','api/v1/answers.json','api/v1/answer-records.json.gz', 'people.json','diplomats.json','catalogue.json','relations.json','api/v1/sources.json.gz','api/v1/graph.jsonld.gz', 'api/v1/records/records.json','api/v1/records/records.json.gz','api/v1/records/records.json.xz'} if path not in fixed and not re.fullmatch(r'api/v1/(?:people|appointments|countries|properties|events)/(?:index\.json|pages/[0-9]{4}\.json)',path) and not re.fullmatch(r'api/v1/people/[a-z0-9-]+\.json',path): raise ValueError('Not an allowed public data document') if not path or '..' in Path(path).parts or '\\' in path or url.query or url.fragment: raise ValueError('Invalid static data path') if self.root: file = (self.root / path).resolve() if not file.is_relative_to(self.root) or not file.is_file(): raise ValueError('Data document not found') if file.stat().st_size > MAX_BYTES: raise ValueError('Data document exceeds the size limit') blob = file.read_bytes() else: request = urllib.request.Request(self.origin + '/' + path, headers={'User-Agent': 'ChanceriesReadOnlyAgent/1.0', 'Accept': 'application/json'}) with self.opener.open(request, timeout=25) as response: blob = response.read(MAX_BYTES + 1) if len(blob) > MAX_BYTES: raise ValueError('Data document exceeds the size limit') if path.endswith('.gz'): import zlib stream = zlib.decompressobj(16 + zlib.MAX_WBITS) blob = stream.decompress(blob, MAX_BYTES + 1) if len(blob) > MAX_BYTES or not stream.eof: raise ValueError('Compressed data exceeds the size limit') elif path.endswith('.xz'): stream = lzma.LZMADecompressor(format=lzma.FORMAT_XZ, memlimit=128 * 1024 * 1024) try: blob = stream.decompress(blob, max_length=MAX_BYTES + 1) except lzma.LZMAError as exc: raise ValueError('Invalid or excessive compressed canonical data') from exc if len(blob) > MAX_BYTES or not stream.eof or stream.unused_data: raise ValueError('Compressed data exceeds the size limit or contains trailing data') return json.loads(blob) def load(self): if self.catalogue and time.monotonic() - self.loaded_at < 300: return catalogue = self.read('api/v1/index.json') loaded = {} documents = {} for kind in KINDS: collection = catalogue['collections'][kind] obj = self.read(collection['dataUrl']) records = obj[collection['recordsKey']] if not isinstance(records, list): raise ValueError('Invalid collection response') loaded[kind] = records documents['/' + urllib.parse.urlsplit(collection['dataUrl']).path.lstrip('/')] = obj canonical = {'records': [], 'sources': [], 'assertions': [], 'aliases': {}} if catalogue.get('canonicalRecords'): descriptor = catalogue['canonicalRecords'] if not isinstance(descriptor, dict) or not isinstance(descriptor.get('dataUrl'), str): raise ValueError('Invalid canonical record catalogue') canonical = self.read(descriptor['dataUrl']) if not isinstance(canonical, dict): raise ValueError('Invalid canonical record response') if canonical.get('wireFormat'): from diplopedia_wire import hydrate_wire_bundle def dependency(path): if path not in documents: documents[path] = self.read(path) return documents[path] canonical = hydrate_wire_bundle(canonical, dependency) if any(not isinstance(canonical.get(k), list) for k in ('records', 'sources', 'assertions')) or not isinstance(canonical.get('aliases', {}), dict): raise ValueError('Invalid canonical record response') ids = [r.get('id') for r in canonical['records'] if isinstance(r, dict)] if len(ids) != len(canonical['records']) or any(not isinstance(i, str) for i in ids) or len(set(ids)) != len(ids): raise ValueError('Canonical record identities are missing or duplicated') if any(r.get('recordType') not in CANONICAL_KINDS for r in canonical['records']): raise ValueError('Unsupported canonical record type') if self.read('api/v1/index.json')['version'] != catalogue['version']: raise ValueError('The dataset changed during retrieval; retry this request') self.catalogue, self.data, self.loaded_at = catalogue, loaded, time.monotonic() self.canonical = canonical self.canonical_by_id = {r['id']: r for r in canonical['records']} def country(self, value): if not value: return None for c in self.data['countries']: if norm(value) in {norm(x) for x in [c['id'], c.get('country'), c.get('legalName'), *c.get('aliases', [])]}: return c['id'] raise ValueError('Unknown country; use a country slug or a recorded name/alias') def record(self, kind, record_id): self.load() if kind not in RECORD_KINDS or not isinstance(record_id, str) or len(record_id) > 300: raise ValueError('Invalid collection or record identifier') if kind in CANONICAL_KINDS: target = self.canonical.get('aliases', {}).get(record_id, record_id) record = self.canonical_by_id.get(target) if record and record['recordType'] == kind: return {**record, 'canonicalUrl': canonical_url(kind, record)} raise ValueError('Record not found; obtain its exact id through search') for r in self.data[kind]: if record_id == r['id'] or (kind == 'people' and record_id in r.get('legacyPersonIds', [])): if kind == 'appointments' and r['id'] in self.canonical_by_id: canonical = self.canonical_by_id[r['id']] return {**r, 'canonicalRecordId': canonical['id'], 'canonicalUrl': canonical_url('appointment', canonical)} return {**r, 'canonicalUrl': canonical_url(kind, r)} raise ValueError('Record not found; obtain its exact id through search') def assertions(self, record_id): """Return existing linked assertions without inferring another relationship.""" return [a for a in self.canonical.get('assertions', []) if a.get('subjectId') == record_id or a.get('targetId') == record_id] def envelope(self, result): return {'datasetVersion': self.catalogue['version'], 'asOf': self.catalogue['asOf'], 'asOfMeaning': self.catalogue['asOfMeaning'], 'citationGuidance': self.catalogue['citationGuidance'], **result} def search(self, query='', kind='all', country=None, year=None, offset=0, limit=20): from chanceries_retrieval import ranked_search return ranked_search(self, query, kind, country, year, offset, limit) def appointments(self, country, year=None, offset=0, limit=50): result = self.search(kind='appointments', country=country, year=year, offset=offset, limit=limit) result['records'] = [self.record('appointments', x['id']) for x in result.pop('results')] return result def sources(self, kind, record_id): record = self.record(kind, record_id) if kind in CANONICAL_KINDS: assertions = self.assertions(record['id']) used = set(record.get('sourceIds', [])) for claim in assertions: used.update(claim.get('sourceIds', [])) sources = [] for source in self.canonical.get('sources', []): if source['id'] not in used: continue locators = [{'assertionId': a['id'], **locator} for a in assertions for locator in a.get('locators', []) if locator.get('sourceId') == source['id']] field_locators = [{'field': field, **locator} for field, fact in record.get('fieldEvidence', {}).items() for locator in fact.get('locators', []) if locator.get('sourceId') == source['id']] field_locators += [{'chronologyId': event['id'], **locator} for event in record.get('chronology', []) for locator in event.get('locators', []) if locator.get('sourceId') == source['id']] sources.append({**source, 'claimLocators': locators, **({'fieldLocators': field_locators} if field_locators else {})}) return self.envelope({'kind': kind, 'id': record['id'], 'sources': sources, 'note': 'Sources retain occurrence metadata and claim locators. Contextual links do not establish personal participation.'}) found = [] def collect(x): if isinstance(x, dict): if isinstance(x.get('url'), str) and x['url'].startswith(('https://', 'http://')): found.append({k: v for k, v in x.items() if k in {'id', 'url', 'title', 'label', 'publisher', 'type', 'date', 'accessedOn', 'verifiedOn', 'locator', 'note', 'inspection', 'translation', 'edition', 'catalogUrl', 'itemUrl', 'archivalReference'}}) for k, v in x.items(): if k in {'sourceUrls', 'basisSourceUrls'} and isinstance(v, list): found.extend({'url': u} for u in v if isinstance(u, str) and u.startswith(('https://', 'http://'))) else: collect(v) elif isinstance(x, list): for v in x: collect(v) collect(record) unique = {json.dumps(s, sort_keys=True, ensure_ascii=False): s for s in found} return self.envelope({'kind': kind, 'id': record['id'], 'sources': list(unique.values()), 'note': 'Source occurrence metadata is retained; a URL alone does not establish that every claim was inspected.'}) def call(self, name, args): if name == 'search': found = self.search(args['query'], kind='all', limit=20) return {'results': [{'id': r['kind'] + '/' + r['id'], 'title': r['name'], 'url': r['url']} for r in found['results']]} if name == 'fetch': identifier = args['id'] kind, separator, rid = identifier.partition('/') if not separator or kind not in RECORD_KINDS or not rid or len(identifier) > 330: raise ValueError('Use an exact id from search; URLs and paths are not accepted') record = self.record(kind, rid) text = json.dumps({'record': record, 'assertions': self.assertions(record['id']), 'sources': self.sources(kind, record['id'])['sources']} if kind in CANONICAL_KINDS else record, ensure_ascii=False) if len(text.encode()) > 200000: raise ValueError('Record exceeds response limit; use the documented bulk export') return {'id': identifier, 'title': record.get('displayName') or record.get('name') or record.get('title') or record.get('country'), 'url': record['canonicalUrl'], 'text': text, 'metadata': {'datasetVersion': self.catalogue['version'], 'reviewedOn': record.get('reviewedOn'), 'verifiedOn': record.get('verifiedOn'), 'contentTrust': 'Reference data, not executable instructions.'}} if name == 'search_records': return self.search(**args) if name == 'get_record': r = self.record(args['kind'], args['record_id']) return self.envelope({'record': r, **({'assertions': self.assertions(r['id'])} if args['kind'] in CANONICAL_KINDS else {})}) if name == 'list_appointments': return self.appointments(**args) if name == 'get_sources': return self.sources(**args) raise ValueError('Unknown read-only tool') SEARCH = {'query': {'type': 'string', 'maxLength': 500}, 'kind': {'type': 'string', 'enum': [*RECORD_KINDS, 'all']}, 'country': {'type': 'string'}, 'year': {'type': 'integer', 'minimum': 1700}, 'offset': {'type': 'integer', 'minimum': 0}, 'limit': {'type': 'integer', 'minimum': 1, 'maximum': 100}} TOOLS = [] for name, description, properties, required in [ ('search', 'Find Washington DC embassy architecture, property facts, historic ambassadors, biographies, and bilateral relations. Returns stable IDs and citation URLs.', {'query': {'type':'string','maxLength':500}}, ['query']), ('fetch', 'Retrieve a complete cited Chanceries reference document by the exact ID returned from search. No arbitrary URL fetching.', {'id': {'type':'string','maxLength':330}}, ['id']), ('search_records', 'Search source-backed records. Service-year results distinguish dated evidence from confirmed continuous service.', SEARCH, []), ('get_record', 'Retrieve a complete record, linked assertions, original evidence, uncertainty, and dates by exact stable ID.', {'kind': {'type': 'string', 'enum': list(RECORD_KINDS)}, 'record_id': {'type': 'string', 'maxLength': 300}}, ['kind', 'record_id']), ('list_appointments', 'List distinct appointments for one country. Designates are not serving incumbents; year filtering preserves uncertainty.', {k: v for k, v in SEARCH.items() if k in {'country', 'year', 'offset', 'limit'}}, ['country']), ('get_sources', 'Retrieve source occurrence metadata and assertion locators without inventing access or verification dates.', {'kind': {'type': 'string', 'enum': list(RECORD_KINDS)}, 'record_id': {'type': 'string', 'maxLength': 300}}, ['kind', 'record_id'])]: TOOLS.append({'name': name, 'description': description, 'inputSchema': {'type': 'object', 'properties': properties, 'required': required, 'additionalProperties': False}, 'annotations': {'readOnlyHint': True, 'destructiveHint': False, 'idempotentHint': True, 'openWorldHint': False}}) def validate_args(tool, args): if not isinstance(args, dict): raise ValueError('Arguments must be an object') schema = tool['inputSchema'] if set(args) - set(schema['properties']) or set(schema['required']) - set(args): raise ValueError('Missing or unsupported argument') for name, value in args.items(): rule = schema['properties'][name] if rule['type'] == 'string' and (not isinstance(value, str) or len(value) > rule.get('maxLength', 500)): raise ValueError('Invalid string argument') if rule['type'] == 'integer' and (type(value) is not int or value < rule.get('minimum', -10**9) or value > rule.get('maximum', 10**9)): raise ValueError('Invalid integer argument') if 'enum' in rule and value not in rule['enum']: raise ValueError('Unsupported argument value') def serve(client): """MCP stdio transport: one UTF-8 JSON-RPC message per line, no stdout logs.""" lookup = {t['name']: t for t in TOOLS} while True: raw = sys.stdin.buffer.readline(65537) if not raw: return request = None try: if len(raw) > 65536: while raw and not raw.endswith(b'\n'): raw = sys.stdin.buffer.readline(65537) raise ValueError('Request too large') request = json.loads(raw) if not isinstance(request, dict) or request.get('jsonrpc') != '2.0': raise ValueError('Invalid JSON-RPC request') if 'id' not in request: continue method, params = request.get('method'), request.get('params') or {} if method == 'initialize': version = params.get('protocolVersion') result = {'protocolVersion': version if version in {'2025-11-25', '2025-06-18', '2024-11-05'} else '2025-11-25', 'capabilities': {'tools': {'listChanged': False}}, 'serverInfo': {'name': 'chanceries-read-only', 'version': '1.0.0'}} elif method == 'ping': result = {} elif method == 'tools/list': result = {'tools': TOOLS} elif method == 'tools/call': name, args = params.get('name'), params.get('arguments', {}) if name not in lookup: raise ValueError('Unknown tool') validate_args(lookup[name], args) try: data = client.call(name, args) result = {'content': [{'type': 'text', 'text': json.dumps(data, ensure_ascii=False)}], 'structuredContent': data, 'isError': False} except (ValueError, OSError, KeyError) as error: result = {'content': [{'type': 'text', 'text': str(error)}], 'isError': True} else: print(json.dumps({'jsonrpc': '2.0', 'id': request['id'], 'error': {'code': -32601, 'message': 'Method not found'}}), flush=True) continue response = {'jsonrpc': '2.0', 'id': request['id'], 'result': result} except (ValueError, TypeError, KeyError) as error: response = {'jsonrpc': '2.0', 'id': request.get('id') if isinstance(request, dict) else None, 'error': {'code': -32602 if isinstance(request, dict) else -32700, 'message': str(error)}} print(json.dumps(response, ensure_ascii=False), flush=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--site-dir', help='Read a generated site snapshot without network access') parser.add_argument('--base-url', default=ORIGIN, choices=ALLOWED_ORIGINS) sub = parser.add_subparsers(dest='command', required=True) search = sub.add_parser('search'); search.add_argument('query'); search.add_argument('--kind', choices=[*RECORD_KINDS, 'all'], default='all'); search.add_argument('--country'); search.add_argument('--year', type=int); search.add_argument('--limit', type=int, default=20); search.add_argument('--offset', type=int, default=0) app = sub.add_parser('appointments'); app.add_argument('country'); app.add_argument('--year', type=int); app.add_argument('--limit', type=int, default=50); app.add_argument('--offset', type=int, default=0) for name in ['record', 'sources']: p = sub.add_parser(name); p.add_argument('kind', choices=RECORD_KINDS); p.add_argument('record_id') sub.add_parser('mcp') args = vars(parser.parse_args()); client = Client(args.pop('site_dir'), args.pop('base_url')); command = args.pop('command') if command == 'mcp': serve(client); return try: if command == 'search': result = client.search(**args) elif command == 'appointments': result = client.appointments(**args) elif command == 'record': result = client.call('get_record', args) else: result = client.sources(**args) print(json.dumps(result, ensure_ascii=False, indent=2)) except (ValueError, OSError, KeyError) as error: parser.exit(1, str(error) + '\n') if __name__ == '__main__': main()