op_row
def op_row(
op
):One operation as a readable row, signed by fastcore.apisurface.
spec
Any documented HTTP API as a tool group: load a spec, read the operations, call one.
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 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 into the link to the raw file, since the page itself is 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 builds the fastspec parser for a document, choosing the Discovery or the OpenAPI reader by the document’s shape, and spec_ops returns the operations, at most MAX_OPS of them. norm_paths first merges Azure’s x-ms-paths into paths. 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 with a message that says what went wrong.
spec_ops
def spec_ops(
spec
):Every operation in spec as OpSpec records, whatever its flavour.
load_spec
def load_spec(
src, timeout:int=30
):Load a JSON or YAML spec from a URL, a path, or a parsed dict.
SpecError
def SpecError(
*args, **kwargs
):A specification could not be read, or an operation could not be called.
parse_ops
def parse_ops(
spec
):A fastspec.SpecParser for an OpenAPI/Azure or Google Discovery document.
Azure Blob Storage shows why 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 on code-host links and the refusal of a web page.
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_segmenttest_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('<!DOCTYPE html><html><body>hi</body></html>', 'that address'),
contains='served a web page')The host
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.
SpecHost
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.
spec_markdown
def spec_markdown(
host, name:str=''
):Render a loaded spec as markdown, ## per group, ### per op: (title, md).
op_markdown
def op_markdown(
r
):One operation: what it does, how to call it, and what it insists on.
op_heading
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.
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 renders one operation and 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.
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)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('###') == 3A SpecHost offers the three api tools, and a plain LocalHost offers none of them:
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:
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.
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')