# spec An agent that has to call an HTTP API with no client library can fetch pages and run commands, but neither tells it what the API accepts. Without a description it has to guess paths and field names, and a wrong guess can come back as a plausible `200` or an unhelpful `400`. Most public APIs publish a specification: a machine-readable document that lists every operation, its HTTP verb and path, and its parameters and their types. This module reads two families of them: OpenAPI, including Swagger 2.0 and Azure’s variant, and Google Discovery documents. `fastspec` parses a document into `OpSpec` records, one per operation, each holding the verb, path, parameters, types, defaults and documentation. `fastcore.apisurface` turns each record into a Python signature such as `list_repo_issues(owner, repo, state='open')`. The result is the `api` tool group, which Ramabana offers when it runs with `--spec`. It has three tools. `api_load` reads a specification, `api_ops` lists its operations with their signatures, and `api_call` calls one of them by name. ## When to use a specification A maintained client library is the better choice when one exists. A specification helps with internal services, clients that have fallen behind their API, and one-off endpoints. The web tools find one: `web_search` for where a service publishes its document, and `read_url` for its documentation pages. ## Why the host holds the specifications A loaded specification stays on the host for the rest of the session, so the agent can browse its operations again without fetching the document again. Because this state lives on the host, the terminal and the editor server share one implementation. ## Reading a document [`load_spec`](https://vedicreader.github.io/ramabana/spec.html#load_spec) takes a URL, a file path, or a dict that is already parsed, and returns the document as a dict. It reads JSON or YAML. A URL is fetched with fossick, which comes with the `[search]` extra. A link to a file’s page on GitHub, GitLab or Bitbucket is first turned by [`raw_url`](https://vedicreader.github.io/ramabana/spec.html#raw_url) into the link to the raw file, since the page itself is HTML. [`parse_spec`](https://vedicreader.github.io/ramabana/spec.html#parse_spec) does the parsing, and when it is handed a web page instead of a document it says so and explains how to get the raw link. [`parse_ops`](https://vedicreader.github.io/ramabana/spec.html#parse_ops) builds the `fastspec` parser for a document, choosing the Discovery or the OpenAPI reader by the document’s shape, and [`spec_ops`](https://vedicreader.github.io/ramabana/spec.html#spec_ops) returns the operations, at most `MAX_OPS` of them. [`norm_paths`](https://vedicreader.github.io/ramabana/spec.html#norm_paths) first merges Azure’s `x-ms-paths` into `paths`. [`op_row`](https://vedicreader.github.io/ramabana/spec.html#op_row) turns one operation into the dict that `api_ops` returns: group, name, verb, path, summary, signature, required parameters and a documentation link. Every failure raises [`SpecError`](https://vedicreader.github.io/ramabana/spec.html#specerror) with a message that says what went wrong. ------------------------------------------------------------------------ source ### op_row ``` python def op_row( op ): ``` *One operation as a readable row, signed by `fastcore.apisurface`.* ------------------------------------------------------------------------ source ### spec_ops ``` python def spec_ops( spec ): ``` *Every operation in `spec` as `OpSpec` records, whatever its flavour.* ------------------------------------------------------------------------ source ### norm_paths ``` python def norm_paths( spec ): ``` *`spec` with `x-ms-paths` folded into `paths`.* ------------------------------------------------------------------------ source ### load_spec ``` python def load_spec( src, timeout:int=30 ): ``` *Load a JSON or YAML spec from a URL, a path, or a parsed dict.* ------------------------------------------------------------------------ source ### raw_url ``` python def raw_url( s ): ``` *The raw file behind a code-host page url.* ------------------------------------------------------------------------ source ### parse_spec ``` python def parse_spec( text, why:str='' ): ``` *A spec document as a dict, from JSON or YAML.* ------------------------------------------------------------------------ source ### SpecError ``` python def SpecError( *args, **kwargs ): ``` *A specification could not be read, or an operation could not be called.* ------------------------------------------------------------------------ source ### parse_ops ``` python def parse_ops( spec ): ``` *A `fastspec.SpecParser` for an OpenAPI/Azure or Google Discovery document.* Azure Blob Storage shows why [`norm_paths`](https://vedicreader.github.io/ramabana/spec.html#norm_paths) exists. Its Swagger 2.0 document leaves `paths` empty and declares its operations under `x-ms-paths`, because Swagger 2.0 cannot list operations that differ only in their query string, such as `/?comp=list` and `/?restype=service&comp=properties`, under `paths`. A parser that reads only `paths` finds no operations at all. `nbs/fixtures/azure-blob-excerpt.json` holds three of the original entries, so the example runs without the network. It finds six operations, and when a path appears in both places the entry under `paths` wins. The cell after it checks [`raw_url`](https://vedicreader.github.io/ramabana/spec.html#raw_url) on code-host links and the refusal of a web page. ``` python azure = load_spec(Path('fixtures/azure-blob-excerpt.json')) assert azure['swagger'] == '2.0' and not azure['paths'], 'the whole point: paths is empty' assert len(azure['x-ms-paths']) == 3 ops = spec_ops(azure) rows = [op_row(o) for o in ops] assert len(rows) == 6, 'three paths, six operations, none of them under `paths`' assert {r['verb'] for r in rows} == {'GET', 'PUT', 'DELETE'} assert any(r['path'] == '/?comp=list' for r in rows), 'a query-only path survived' assert all(r['signature'].startswith(r['name']) for r in rows) # `paths` wins a collision clash = dict(azure, paths={'/?comp=list': {'get': {'operationId': 'mine'}}}) assert norm_paths(clash)['paths']['/?comp=list'] == {'get': {'operationId': 'mine'}} for r in rows[:3]: print(f"{r['verb']:6} {r['path'][:38]:40} {r['name']}") ``` PUT /?restype=service&comp=properties service__set_properties GET /?restype=service&comp=properties service__get_properties GET /?comp=list service__list_containers_segment ``` python test_eq(raw_url('https://github.com/stripe/openapi/blob/master/openapi/spec3.json'), 'https://raw.githubusercontent.com/stripe/openapi/master/openapi/spec3.json') test_eq(raw_url('https://github.com/o/r/blob/main/a/b.yaml?plain=1#L4'), 'https://raw.githubusercontent.com/o/r/main/a/b.yaml') test_eq(raw_url('https://gitlab.com/o/r/-/blob/main/api.yaml'), 'https://gitlab.com/o/r/-/raw/main/api.yaml') for u in ('https://raw.githubusercontent.com/o/r/main/s.json', 'https://api.example.com/openapi.json', 'https://github.com/o/r', 'specs/local.yaml'): test_eq(raw_url(u), u) test_fail(lambda: parse_spec('hi', 'that address'), contains='served a web page') ``` ## The host [`SpecHost`](https://vedicreader.github.io/ramabana/spec.html#spechost) is a `LocalHost` that also holds specifications, by name, and it is what makes the `api` tool group available. `api_load(src, name)` reads a document and keeps it under `name`, or under its title when no name is given. It returns a short summary: the operation count, the groups, and a note when the list is longer than one page. `api_names` lists what is loaded. `api_ops` lists the operations, narrowed by `group` or by `match`, a piece of text to look for in the name or summary, and paged with `limit` and `offset`. `api_count` says how many operations a narrowing matches in all. `api_call(operation, **params)` calls one operation by its name, with the parameter names its signature shows. When more than one specification is loaded, each of these needs `name` to say which one. `headers` go with every call to every specification. `api_creds` sets headers, such as an API key, for one specification only, as the last example shows. ------------------------------------------------------------------------ source ### SpecHost ``` python def SpecHost( roots:tuple=('.',), # the folders the agent is confined to specs:NoneType=None, # name -> parsed spec, preloaded headers:NoneType=None, # sent with every API call creds:NoneType=None, # name -> headers, sent only to that spec timeout:float=60.0, # per-call timeout, in seconds max_ops:int=400, # rows per `api_ops` page; 0 for all *, ns:NoneType=None, # live namespace; fresh dict when None approvals:NoneType=None, # write approval handler note:NoneType=None, # status callback web:bool=True, # use fossick when installed index:bool=True, # start a Kosha sync for every open root graph:bool=False, # build Kosha's call graph during sync rerank:bool=True, # rerank Kosha hits with flashrank rerank_model:NoneType=None, # flashrank model; None uses its default memory:NoneType=None, # memory, ask and watch backend apis:NoneType=None, # API specification backend kernel:NoneType=None, # live kernel; None runs in process read_outside:bool=False, # let read-only tools name any path on this machine deny:tuple=('*/.ssh/*', '*/.aws/*', '*/.gnupg/*', '*/.config/gcloud/*', '*/.netrc', '*/.git-credentials', '*/.codex/auth.json', '*/.claude/.credentials.json', '*/.env', '*/.env.*', '*/id_rsa*', '*/id_ed25519*', '*.pem', '*.key', '*.p12'), # what `read_outside` still refuses to open read_roots:tuple=(), # folders the read tools search and no write may touch tmux:NoneType=None, # None detects `$TMUX`; False off; True on; or this host's own pane (target or `Pane`) read_guard:NoneType=None, # `(ReadCheck) -> None | str | Refusal`: what the read tools may send model_local:NoneType=None, # `() -> (local, model)` for the turn's model ): ``` *`LocalHost` plus named API specifications (OpenAPI or Google Discovery), loaded on demand.* ------------------------------------------------------------------------ source ### spec_markdown ``` python def spec_markdown( host, name:str='' ): ``` *Render a loaded spec as markdown, `##` per group, `###` per op: `(title, md)`.* ------------------------------------------------------------------------ source ### op_markdown ``` python def op_markdown( r ): ``` *One operation: what it does, how to call it, and what it insists on.* ------------------------------------------------------------------------ source ### op_heading ``` python def op_heading( r ): ``` *`POST /widgets/{id}`: unique per operation, readable without the spec.* ## Examples The examples build small specifications in the notebook, so none of them needs the network. The first loads a two-operation OpenAPI document, lists its operations with the signature `api_ops` gives each one, and shows the errors when `name` is missing or names nothing loaded. ``` python SPEC = {'openapi': '3.0.0', 'info': {'title': 'Widgets', 'version': '1.2.0'}, 'paths': {'/widgets': {'get': {'operationId': 'listWidgets', 'tags': ['widgets'], 'summary': 'Every widget', 'parameters': [{'name': 'limit', 'in': 'query', 'schema': {'type': 'integer'}}]}}, '/widgets/{id}': {'get': {'operationId': 'getWidget', 'tags': ['widgets'], 'summary': 'One widget', 'parameters': [{'name': 'id', 'in': 'path', 'required': True, 'schema': {'type': 'string'}}]}}}} h = SpecHost(roots=['.']) loaded = h.api_load(SPEC, 'widgets') test_eq(loaded['operations'], 2) test_eq(loaded['title'], 'Widgets') rows = {r['name']: r for r in h.api_ops(name='widgets')} test_eq(sorted(rows), ['get_widget', 'list_widgets']) assert 'id' in rows['get_widget']['signature'], rows['get_widget']['signature'] assert 'limit' in rows['list_widgets']['signature'], rows['list_widgets']['signature'] test_eq(rows['get_widget']['verb'], 'GET') test_eq(rows['get_widget']['required'], ['id']) test_eq([r['name'] for r in h.api_ops(match='one widget', name='widgets')], ['get_widget']) test_eq(h.api_ops(group='nothing', name='widgets'), []) h.api_load(SPEC, 'other') test_fail(lambda: h.api_ops(), contains='name which api') test_fail(lambda: h.api_ops(name='nope'), contains='no api') print(rows['get_widget']['signature']) ``` get_widget(id: str) ### A specification as a document A real specification can hold hundreds of operations, too many to read as a list of rows. `spec_markdown(host, name)` renders a loaded one as a reference page in markdown and returns `(title, markdown)`. Each group gets a `##` heading and each operation a `###` heading, followed by what the operation does, how to call it and which parameters it requires. It uses the same rows as `api_ops`, all of them rather than one page. [`op_markdown`](https://vedicreader.github.io/ramabana/spec.html#op_markdown) renders one operation and [`op_heading`](https://vedicreader.github.io/ramabana/spec.html#op_heading) writes its heading. Operation names are often lowercase, and some markdown readers handle lowercase headings badly, so group headings start with a capital letter. Each operation is headed by its verb and path, which stays unique when several operations share a path. The parsed document keeps only its base URL and its operations, so `api_load` saves the title, version and source in `spec_info` for this page to use. The first cell below loads a 25-operation document into a host that shows ten per page, and checks the paging and narrowing. The second renders a document with two groups. ``` python BIG = {'openapi': '3.0.0', 'info': {'title': 'Big', 'version': '1'}, 'paths': {f'/g{i//10}/thing{i}': {'get': {'operationId': f'getThing{i}', 'summary': 'the last one' if i == 24 else f'thing {i}'}} for i in range(25)}} small = SpecHost(roots=['.'], max_ops=10) info = small.api_load(BIG, 'big') test_eq(info['operations'], 25) test_eq(info['shown'], 10) assert info['truncated'] and 'limit' in info['more'], info test_eq(len(small.api_ops(name='big')), 10) test_eq(len(small.api_ops(name='big', limit=0)), 25) test_eq(len(small.api_ops(name='big', limit=5, offset=22)), 3) test_eq(small.api_count(name='big'), 25) test_eq(info['groups'], ['g0', 'g1', 'g2']) test_eq([r['name'] for r in small.api_ops(name='big', match='the last one')], ['get_thing24']) test_eq(small.api_count(name='big', group='g2'), 5) whole = SpecHost(roots=['.'], max_ops=0) assert 'truncated' not in whole.api_load(BIG, 'big') test_eq(len(whole.api_ops(name='big')), 25) ``` ``` python ADMIN = {'openapi': '3.0.0', 'info': {'title': 'Widgets', 'version': '1.2.0'}, 'paths': {**SPEC['paths'], '/admin/flush': {'post': {'operationId': 'flush', 'tags': ['widgets'], 'summary': 'Empty the cache'}}}} d = SpecHost(roots=['.']) d.api_load(ADMIN, 'widgets') assert not getattr(d.specs['widgets'], 'title', None) test_eq(d.spec_info['widgets']['title'], 'Widgets') test_eq(d.spec_info['widgets']['version'], '1.2.0') test_eq(d.spec_info['widgets']['url'], '') test_eq(SpecHost(roots=['.']).spec_info, {}) title, doc = spec_markdown(d, 'widgets') test_eq(title, 'Widgets') assert doc.startswith('3 operations in version 1.2.0'), doc[:60] assert '## Admin' in doc and '## Widgets' in doc assert doc.index('## Admin') < doc.index('## Widgets') assert '### GET /widgets/{id}' in doc and '### POST /admin/flush' in doc assert 'Call it as `get_widget(' in doc and 'Required: id' in doc assert not doc.startswith('#') and '\n# ' not in doc heads = [l for l in doc.splitlines() if l.startswith('#')] assert heads and all(l.lstrip('#').strip()[0].isupper() for l in heads), heads d.spec_info.clear() test_eq(spec_markdown(d, 'widgets')[0], 'Widgets') test_eq(op_heading({'verb': 'delete', 'path': '/widgets/{id}'}), 'DELETE /widgets/{id}') bare = op_markdown({'verb': 'get', 'path': '/x', 'name': 'x'}) assert bare.startswith('### GET /x') and 'Required' not in bare and 'Docs' not in bare d.api_load(ADMIN, 'second') test_fail(lambda: spec_markdown(d, ''), contains='name which api') print(doc) paged = SpecHost(roots=['.'], max_ops=1) paged.api_load(ADMIN, 'widgets') assert spec_markdown(paged, 'widgets')[1].count('###') == 3 ``` A [`SpecHost`](https://vedicreader.github.io/ramabana/spec.html#spechost) offers the three `api` tools, and a plain `LocalHost` offers none of them: ``` python from ramabana.tools import tools_for names = {t.__name__ for t in tools_for(SpecHost(roots=['.']))} assert {'api_load', 'api_ops', 'api_call'} <= names, sorted(names) assert not {'api_load', 'api_ops', 'api_call'} & {t.__name__ for t in tools_for(LocalHost(roots=['.']))} print(sorted(n for n in names if n.startswith('api_'))) ``` ['api_call', 'api_load', 'api_ops'] A Google Discovery document loads the same way. Its operations are grouped by resource, here `buckets`, and its base URL comes from `baseUrl`: ``` python GCP = { 'kind': 'discovery#restDescription', 'discoveryVersion': 'v1', 'name': 'storage', 'version': 'v1', 'title': 'Cloud Storage JSON API', 'baseUrl': 'https://storage.googleapis.com/storage/v1/', 'resources': {'buckets': {'methods': {'get': { 'id': 'storage.buckets.get', 'path': 'b/{bucket}', 'httpMethod': 'GET', 'parameters': {'bucket': {'location': 'path', 'required': True, 'type': 'string'}, 'projection': {'location': 'query', 'type': 'string'}}, 'description': 'Returns metadata for the specified bucket.' }}}} } gcp = SpecHost(roots=['.']) loaded_gcp = gcp.api_load(GCP, 'gcp') test_eq(loaded_gcp['title'], 'Cloud Storage JSON API') test_eq(loaded_gcp['operations'], 1) gcp_row = gcp.api_ops(name='gcp')[0] test_eq((gcp_row['group'], gcp_row['name'], gcp_row['path'], gcp_row['required']), ('buckets', 'get', 'b/{bucket}', ['bucket'])) test_eq(gcp.specs['gcp'].base_url, 'https://storage.googleapis.com/storage/v1/') from fastspec.oapi import OpenAPIClient assert isinstance(OpenAPIClient(gcp.specs['gcp'], sync=True), OpenAPIClient) ``` ### A key for one specification Headers passed as `headers` go with every call. Credentials should not: one host can hold several specifications, and an `Authorization` header meant for one service would then reach all the others. `api_creds(name, headers)` stores headers for one specification only, and calling it with `headers=None` removes them. `api_keyed` shows which specifications carry credentials and the names of their headers, never the values. In the example only the client for `rates` sends the API key. ``` python RATES = {'openapi': '3.0.0', 'info': {'title': 'Rates', 'version': '1'}, 'servers': [{'url': 'https://api.example.com'}], 'paths': {'/v1/latest': {'get': {'operationId': 'latest', 'summary': 'Latest rates'}}}} k = SpecHost(roots=['.'], headers={'User-Agent': 'ramabana'}) k.api_load(RATES, 'rates'); k.api_load(SPEC, 'widgets') k.api_creds('rates', {'X-Api-Key': 'secret'}) test_eq(k.api_keyed(), {'rates': ['X-Api-Key']}) test_eq(k._client('rates', k.specs['rates']).transport.base_headers, {'User-Agent': 'ramabana', 'X-Api-Key': 'secret'}) test_eq(k._client('widgets', k.specs['widgets']).transport.base_headers, {'User-Agent': 'ramabana'}) assert not callable(getattr(k._client('rates', k.specs['rates']), 'latest')) test_fail(lambda: k.api_call('nope', name='rates'), contains='no operation') ```