funccall source

Function calling

Function to schema

Many LLMs do function calling (aka tool use) by taking advantage of JSON schema.

We’ll use docments to make getting JSON schema from Python functions as ergonomic as possible. Each parameter (and the return value) should have a type, and a docments comment with the description of what it is. Here’s an example:

def silly_sum(
    a:int, # First thing to sum
    b:int=1, # Second thing to sum
    c:list[int]=None, # A pointless argument
) -> int: # The sum of the inputs
    "Adds a + b."
    return a + b

This is what docments makes of that:

d = docments(silly_sum, full=True)
d

{ 'a': { 'anno': <class 'int'>,
         'default': <class 'inspect._empty'>,
         'docment': 'First thing to sum'},
  'b': {'anno': <class 'int'>, 'default': 1, 'docment': 'Second thing to sum'},
  'c': {'anno': list[int], 'default': None, 'docment': 'A pointless argument'},
  'return': { 'anno': <class 'int'>,
              'default': <class 'inspect._empty'>,
              'docment': 'The sum of the inputs'}}

Note that this is an AttrDict so we can treat it like an object, or a dict:

d.a.docment, d['a']['anno']

('First thing to sum', int)

This private function converts a key/value pair from the docments structure into the dict that will be needed for the schema.

n,o = first(d.items())
print(n,'//', o)
_param(o)

a // {'docment': 'First thing to sum', 'anno': <class 'int'>, 'default': <class 'inspect._empty'>}

{'description': 'First thing to sum'}

n,o

('a',
 {'docment': 'First thing to sum', 'anno': int, 'default': inspect._empty})

{ 'a': { 'anno': <class 'int'>,
         'default': <class 'inspect._empty'>,
         'docment': 'First thing to sum'},
  'b': {'anno': <class 'int'>, 'default': 1, 'docment': 'Second thing to sum'},
  'c': {'anno': list[int], 'default': None, 'docment': 'A pointless argument'},
  'return': { 'anno': <class 'int'>,
              'default': <class 'inspect._empty'>,
              'docment': 'The sum of the inputs'}}

_handle_type(int, None), _handle_type(Path, None)

({'type': 'integer'}, {'type': 'string', 'format': 'Path'})

Fixed-length tuples (e.g. tuple[int, str]) are mapped to a JSON Schema array with prefixItems for per-position types, which is accepted by Anthropic. OpenAI/Gemini require items so this is added, along with minItems/maxItems to enforce the fixed length. If all positions share the same type, items is that type directly; otherwise it’s an anyOf of the unique types.

# gemini expect `items` to be defined for arrays
_handle_type(list, None), _handle_type(tuple[str], None), _handle_type(set[str], None)

({'type': 'array', 'items': {}},
 {'type': 'array',
  'prefixItems': [{'type': 'string'}],
  'items': {'type': 'string'},
  'minItems': 1,
  'maxItems': 1},
 {'type': 'array', 'items': {'type': 'string'}, 'uniqueItems': True})

_handle_type(dict, None), _handle_type(dict[str,str], None)

({'type': 'object'},
 {'type': 'object', 'additionalProperties': {'type': 'string'}})

def _example_new_unioin(opt_tup: str | None): pass

d = docments(_example_new_unioin, full=True)
anno1 = first(d.items())[1].anno
(anno1, get_origin(anno1), get_args(anno1))

(str | None, types.UnionType, (str, NoneType))

def _example_old_union(opt_tup: Union[str, type(None)] =None): pass

d = docments(_example_old_union, full=True)
anno2 = first(d.items())[1].anno
(anno2, get_origin(anno2), get_args(anno2))

(typing.Optional[str], typing.Union, (str, NoneType))

# Test primitive types
defs = {}
assert _handle_type(int, defs) == {'type': 'integer'}
assert _handle_type(str, defs) == {'type': 'string'}
assert _handle_type(bool, defs) == {'type': 'boolean'}
assert _handle_type(float, defs) == {'type': 'number'}

# Test custom class
class TestClass:
    def __init__(self, x: int, y: int): store_attr()

result = _handle_type(TestClass, defs)
assert result == {'$ref': '#/$defs/TestClass'}
assert 'TestClass' in defs
assert defs['TestClass']['type'] == 'object'
assert 'properties' in defs['TestClass']

# tuple[int, ...] should produce array with items, not prefixItems
test_eq(_handle_type(tuple[int, ...], {}), {'type': 'array', 'items': {'type': 'integer'}})

# Test primitive types in containers
test_eq(_handle_type(list[int], defs), {'type': 'array', 'items': {'type': 'integer'}})
test_eq(_handle_type(tuple[str], defs), {'type': 'array', 'prefixItems': [{'type': 'string'}], 'items': {'type': 'string'}, 'minItems': 1, 'maxItems': 1})
test_eq(_handle_type(set[str], defs), dict(type='array', items={'type': 'string'}, uniqueItems=True))
test_eq(_handle_type(dict[str,bool], defs), {'type': 'object', 'additionalProperties': {'type': 'boolean'}})

result = _handle_type(list[TestClass], defs)
assert result == {'type': 'array', 'items': {'$ref': '#/$defs/TestClass'}}
assert 'TestClass' in defs

# Test complex nested structure
ComplexType = dict[str, list[TestClass]]
result = _handle_type(dict[str, list[TestClass]], defs)
assert result == {'type': 'object', 'additionalProperties': {'type': 'array', 'items': {'$ref': '#/$defs/TestClass'}}}

# Test processing of a required integer property
props, req = {}, {}
class TestClass:
    "Test class"
    def __init__(
        self,
        x: int, # First thing
        y: list[float], # Second thing
        z: str = "default", # Third thing
    ): store_attr()

d = docments(TestClass, full=True)
_process_property('x', d.x, props, req, defs)
assert 'x' in props
assert props['x']['type'] == 'integer'
assert 'x' in req

# Test processing of a required list property
_process_property('y', d.y, props, req, defs)
assert 'y' in props
assert props['y']['type'] == 'array'
assert props['y']['items']['type'] == 'number'
assert 'y' in req

# Test processing of an optional string property with default
_process_property('z', d.z, props, req, defs)
assert 'z' in props
assert props['z']['type'] == 'string'
assert props['z']['default'] == "default"
assert 'z' not in req

source

get_schema


def get_schema(
    f:Union, # Function to get schema for
    pname:str='input_schema', # Key name for parameters
    evalable:bool=False, # stringify defaults that can't be literal_eval'd?
    skip_hidden:bool=False, # skip parameters starting with '_'?
    name:NoneType=None, # Override function name (useful for dotted paths like 'obj.method')
)->dict: # {'name':..., 'description':..., pname:...}

Generate JSON schema for a class, function, or method

get_schema(get_schema)

{'name': 'get_schema',
 'description': "Generate JSON schema for a class, function, or method\n\nReturns:\n- {'name':..., 'description':..., pname:...} (type: object)",
 'input_schema': {'type': 'object',
  'properties': {'f': {'description': 'Function to get schema for',
    'anyOf': [{'type': 'object'}, {'type': 'object'}]},
   'pname': {'description': 'Key name for parameters',
    'default': 'input_schema',
    'type': 'string'},
   'evalable': {'description': "stringify defaults that can't be literal_eval'd?",
    'default': False,
    'type': 'boolean'},
   'skip_hidden': {'description': "skip parameters starting with '_'?",
    'default': False,
    'type': 'boolean'},
   'name': {'description': "Override function name (useful for dotted paths like 'obj.method')",
    'default': None,
    'type': 'null'}},
  'required': ['f']}}

def f(
    o:object, # the o
    q:tuple[int,str],
    p:str|list[str] = 'a',
): "object function"

s = get_schema(f)
test_eq(s['name'], 'f')
inpp = s['input_schema']['properties']
test_eq(inpp['o'], {'type': 'object', 'description': 'the o'})
test_eq(inpp['q'], dict(type='array', description='', prefixItems=[{'type': 'integer'}, {'type': 'string'}], items=dict(anyOf=[{'type': 'integer'}, {'type': 'string'}]), minItems=2, maxItems=2))
test_eq(inpp['p'], dict(description='', default='a', anyOf=[{'type': 'string'}, {'type': 'array', 'items': {'type': 'string'}}]))
s

{'name': 'f',
 'description': 'object function',
 'input_schema': {'type': 'object',
  'properties': {'o': {'description': 'the o', 'type': 'object'},
   'q': {'description': '',
    'type': 'array',
    'prefixItems': [{'type': 'integer'}, {'type': 'string'}],
    'items': {'anyOf': [{'type': 'integer'}, {'type': 'string'}]},
    'minItems': 2,
    'maxItems': 2},
   'p': {'description': '',
    'default': 'a',
    'anyOf': [{'type': 'string'},
     {'type': 'array', 'items': {'type': 'string'}}]}},
  'required': ['o', 'q']}}

class ClassA:
    "I am a class"
    def f(self, a:int): # That is `a`
        "Do a thing"
        return 1
    def __call__(self, b:str): # That is `b`
        "Do another thing"
        return 2

ca = ClassA()
ca.f(2)

get_schema(ca.f)

{'name': 'f',
 'description': 'Do a thing',
 'input_schema': {'type': 'object',
  'properties': {'a': {'description': 'That is `a`', 'type': 'integer'}},
  'required': ['a']}}

get_schema(ca)

{'name': '__call__',
 'description': 'Do another thing',
 'input_schema': {'type': 'object',
  'properties': {'b': {'description': 'That is `b`', 'type': 'string'}},
  'required': ['b']}}

Usage examples

Putting this all together, we can now test getting a schema from silly_sum. The tool use spec doesn’t support return annotations directly, so we put that in the description instead.

s = get_schema(silly_sum)
desc = s.pop('description')
print(desc)
s

Adds a + b.

Returns:
- The sum of the inputs (type: integer)

{'name': 'silly_sum',
 'input_schema': {'type': 'object',
  'properties': {'a': {'description': 'First thing to sum', 'type': 'integer'},
   'b': {'description': 'Second thing to sum',
    'default': 1,
    'type': 'integer'},
   'c': {'description': 'A pointless argument',
    'default': None,
    'type': 'array',
    'items': {'type': 'integer'}}},
  'required': ['a']}}

This also works with string annotations, e.g:

def silly_test(
    a: 'int',  # quoted type hint
)->int:
    "Mandatory docstring"
    return a

get_schema(silly_test)

{'name': 'silly_test',
 'description': 'Mandatory docstring\n\nReturns:\n- type: integer',
 'input_schema': {'type': 'object',
  'properties': {'a': {'description': 'quoted type hint', 'type': 'integer'}},
  'required': ['a']}}

This also works with instance methods:

class Dummy:
    def sums(
        self,
        a:int,  # First thing to sum
        b:int=1 # Second thing to sum
    ): # The sum of the inputs
        "Adds a + b."
        print(f"Finding the sum of {a} and {b}")
        return a + b

get_schema(Dummy.sums)

{'name': 'sums',
 'description': 'Adds a + b.',
 'input_schema': {'type': 'object',
  'properties': {'a': {'description': 'First thing to sum', 'type': 'integer'},
   'b': {'description': 'Second thing to sum',
    'default': 1,
    'type': 'integer'}},
  'required': ['a']}}

get_schema also handles more complicated structures such as nested classes. This is useful for things like structured outputs.

class Turn:
    "Turn between two speakers"
    def __init__(
        self,
        speaker_a:str, # First speaker's message
        speaker_b:str,  # Second speaker's message
    ): store_attr()

class Conversation:
    "A conversation between two speakers"
    def __init__(
        self,
        turns:list[Turn], # Turns of the conversation
    ): store_attr()

get_schema(Conversation)

{'name': 'Conversation',
 'description': 'A conversation between two speakers',
 'input_schema': {'type': 'object',
  'properties': {'turns': {'description': 'Turns of the conversation',
    'type': 'array',
    'items': {'$ref': '#/$defs/Turn'}}},
  'title': 'Conversation',
  'required': ['turns'],
  '$defs': {'Turn': {'type': 'object',
    'properties': {'speaker_a': {'description': "First speaker's message",
      'type': 'string'},
     'speaker_b': {'description': "Second speaker's message",
      'type': 'string'}},
    'title': 'Turn',
    'required': ['speaker_a', 'speaker_b']}}}}

class DictConversation:
    "A conversation between two speakers"
    def __init__(
        self,
        turns:dict[str,object], # dictionary of topics and the Turns of the conversation
    ): store_attr()

get_schema(DictConversation)

{'name': 'DictConversation',
 'description': 'A conversation between two speakers',
 'input_schema': {'type': 'object',
  'properties': {'turns': {'description': 'dictionary of topics and the Turns of the conversation',
    'type': 'object',
    'additionalProperties': {'type': 'object'}}},
  'title': 'DictConversation',
  'required': ['turns']}}

class SetConversation:
    "A conversation between two speakers"
    def __init__(
        self,
        turns:set[Turn], # the unique Turns of the conversation
    ): store_attr()

get_schema(SetConversation)

{'name': 'SetConversation',
 'description': 'A conversation between two speakers',
 'input_schema': {'type': 'object',
  'properties': {'turns': {'description': 'the unique Turns of the conversation',
    'type': 'array',
    'items': {'$ref': '#/$defs/Turn'},
    'uniqueItems': True}},
  'title': 'SetConversation',
  'required': ['turns'],
  '$defs': {'Turn': {'type': 'object',
    'properties': {'speaker_a': {'description': "First speaker's message",
      'type': 'string'},
     'speaker_b': {'description': "Second speaker's message",
      'type': 'string'}},
    'title': 'Turn',
    'required': ['speaker_a', 'speaker_b']}}}}

Additional `get_schema()` Test Cases

Union types are represented in JSON Schema using anyOf.

IntPair = tuple[int, int]
def _union_test(opt_tup: Union[IntPair, str, int]=None):
    "Mandatory docstring"
    return ""
get_schema(_union_test)

{'name': '_union_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'opt_tup': {'description': '',
    'default': None,
    'anyOf': [{'type': 'array',
      'prefixItems': [{'type': 'integer'}, {'type': 'integer'}],
      'items': {'type': 'integer'},
      'minItems': 2,
      'maxItems': 2},
     {'type': 'string'},
     {'type': 'integer'}]}}}}

The new (Python 3.10+) union syntax can also be used, producing an equivalent schema.

def _new_union_test(opt_tup: IntPair | str | int =None):
    "Mandatory docstring"
    pass
get_schema(_new_union_test)

{'name': '_new_union_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'opt_tup': {'description': '',
    'default': None,
    'anyOf': [{'type': 'array',
      'prefixItems': [{'type': 'integer'}, {'type': 'integer'}],
      'items': {'type': 'integer'},
      'minItems': 2,
      'maxItems': 2},
     {'type': 'string'},
     {'type': 'integer'}]}}}}

Optional is a special case of union types, limited to two types, one of which is None (mapped to null in JSON schema):

def _optional_test(opt_tup: Optional[IntPair]=None):
    "Mandatory docstring"
    pass
get_schema(_optional_test)

{'name': '_optional_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'opt_tup': {'description': '',
    'default': None,
    'anyOf': [{'type': 'array',
      'prefixItems': [{'type': 'integer'}, {'type': 'integer'}],
      'items': {'type': 'integer'},
      'minItems': 2,
      'maxItems': 2},
     {'type': 'null'}]}}}}

def _param_union_test(items: list[str] | None = None):
    "Test parameterized container in union"
    pass
get_schema(_param_union_test)

{'name': '_param_union_test',
 'description': 'Test parameterized container in union',
 'input_schema': {'type': 'object',
  'properties': {'items': {'description': '',
    'default': None,
    'anyOf': [{'type': 'array', 'items': {'type': 'string'}},
     {'type': 'null'}]}}}}

Containers can also be used, both in their parameterized form (List[int]) or as their unparameterized raw type (List). In the latter case, the item type is mapped to object in JSON schema.

def _list_test(l: List[int]):
    "Mandatory docstring"
    pass
get_schema(_list_test)

{'name': '_list_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'l': {'description': '',
    'type': 'array',
    'items': {'type': 'integer'}}},
  'required': ['l']}}

def _raw_list_test(l: List):
    "Mandatory docstring"
    pass
get_schema(_raw_list_test)

{'name': '_raw_list_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'l': {'description': '', 'type': 'array', 'items': {}}},
  'required': ['l']}}

The same applies to dictionary, which can similarly be parameterized with key/value types or specified as a raw type.

def _dict_test(d: Dict[str, int]):
    "Mandatory docstring"
    pass
get_schema(_dict_test)

{'name': '_dict_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'d': {'description': '',
    'type': 'object',
    'additionalProperties': {'type': 'integer'}}},
  'required': ['d']}}

def _raw_dict_test(d: Dict):
    "Mandatory docstring"
get_schema(_raw_dict_test)

{'name': '_raw_dict_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'d': {'description': '', 'type': 'object'}},
  'required': ['d']}}

def _path_test(path: Path = Path('.')):
    "Mandatory docstring"
get_schema(_path_test)

{'name': '_path_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'path': {'description': '',
    'default': Path('.'),
    'type': 'string',
    'format': 'Path'}}}}

Schemas that need to be converted using ast.literal_eval will fail with non-primitive defaults:

test_fail(lambda: ast.literal_eval(str(get_schema(_path_test))), exc=ValueError)

Use evalable to have those defaults stringified:

def _path_test(path: Path = Path('.')):
    "Mandatory docstring"
get_schema(_path_test, evalable=True)

{'name': '_path_test',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'path': {'description': '',
    'default': '.',
    'type': 'string',
    'format': 'Path'}}}}

Use skip_hidden to exclude parameters starting with _ from the schema:

def test_hidden(a: int, _internal: str = "x"):
    "Test func"
    pass

get_schema(test_hidden, skip_hidden=True)  # should exclude _internal

{'name': 'test_hidden',
 'description': 'Test func',
 'input_schema': {'type': 'object',
  'properties': {'a': {'description': '', 'type': 'integer'}},
  'required': ['a']}}

get_schema(test_hidden)

{'name': 'test_hidden',
 'description': 'Test func',
 'input_schema': {'type': 'object',
  'properties': {'a': {'description': '', 'type': 'integer'},
   '_internal': {'description': '', 'default': 'x', 'type': 'string'}},
  'required': ['a']}}

def _ret_list_test(a: int) -> list[str]:
    "Mandatory docstring"

s = get_schema(_ret_list_test)
assert 'type: array[string]' in s['description']

Cmd = str | list[str]

def _cust_type(a: Cmd): "Mandatory docstring"

s = get_schema(_cust_type)
s

{'name': '_cust_type',
 'description': 'Mandatory docstring',
 'input_schema': {'type': 'object',
  'properties': {'a': {'description': '',
    'anyOf': [{'type': 'string'},
     {'type': 'array', 'items': {'type': 'string'}}]}},
  'required': ['a']}}

Python tool

In language model clients it’s often useful to have a ‘code interpreter’ – this is something that runs code, and generally outputs the result of the last expression (i.e like IPython or Jupyter).

In this section we’ll create the python function, which executes a string as Python code, with an optional timeout. If the last line is an expression, we’ll return that – just like in IPython or Jupyter, but without needing them installed.

This is an internal function that’s needed for _run to ensure that location information is available in the abstract syntax tree (AST), since otherwise python complains.

This is the internal function used to actually run the code – we pull off the last AST to see if it’s an expression (i.e something that returns a value), and if so, we store it to a special _result variable so we can return it.

_run('import math;math.factorial(12)')

479001600

_run('print(1+1)')

'2'

We now have the machinery needed to create our python function.

source

python


def python(
    code:str, # Code to execute
    glb:Optional=None, # Globals namespace
    loc:Optional=None, # Locals namespace
    timeout:int=3600, # Maximum run time in seconds
):

Executes python code with timeout and returning final expression (similar to IPython).

There’s no builtin security here – you should generally use this in a sandbox, or alternatively prompt before running code. It can handle multiline function definitions, and pretty much any other normal Python syntax.

python("""def factorial(n):
    if n == 0 or n == 1: return 1
    else: return n * factorial(n-1)
factorial(5)""")

If the code takes longer than timeout then it returns an error string.

print(python('import time; time.sleep(10)', timeout=1))

Traceback (most recent call last):
  File "/var/folders/wm/y9k35r7n7q56mvx2wnndd0880000gp/T/ipykernel_61561/3422083997.py", line 13, in python
    try: return _run(code, glb, loc)
                ^^^^^^^^^^^^^^^^^^^^
  File "/var/folders/wm/y9k35r7n7q56mvx2wnndd0880000gp/T/ipykernel_61561/2387754473.py", line 17, in _run
    try: exec(compiled_code, glb, loc)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<ast>", line 1, in <module>
  File "/var/folders/wm/y9k35r7n7q56mvx2wnndd0880000gp/T/ipykernel_61561/3422083997.py", line 8, in handler
    def handler(*args): raise TimeoutError()
                        ^^^^^^^^^^^^^^^^^^^^
TimeoutError

By default the caller’s global namespace is used.

python("a=1")
a

Pass a different glb if needed; this requires using python_ns.

glb = {}
python("a=3", glb=glb)
a, glb['a']

(1, 3)

get_schema(python)

{'name': 'python',
 'description': 'Executes python `code` with `timeout` and returning final expression (similar to IPython).',
 'input_schema': {'type': 'object',
  'properties': {'code': {'description': 'Code to execute', 'type': 'string'},
   'glb': {'description': 'Globals namespace',
    'default': None,
    'anyOf': [{'type': 'object'}, {'type': 'null'}]},
   'loc': {'description': 'Locals namespace',
    'default': None,
    'anyOf': [{'type': 'object'}, {'type': 'null'}]},
   'timeout': {'description': 'Maximum run time in seconds',
    'default': 3600,
    'type': 'integer'}},
  'required': ['code']}}

Tool Calling

Many LLM API providers offer tool calling where an LLM can choose to call a given tool. This is also helpful for structured outputs since the response from the LLM is contrained to the required arguments of the tool.

This section will be dedicated to helper functions for calling tools. We don’t want to allow LLMs to call just any possible function (that would be a security disaster!) so we create a namespace – that is, a dictionary of allowable function names to call.

source

mk_ns


def mk_ns(
    fs
):

Call self as a function.

def sums(a, b): return a + b
ns = mk_ns(sums)
ns

{'sums': <function __main__.sums(a, b)>}

ns['sums'](1, 2)

ca = ClassA()

source

resolve_nm


def resolve_nm(
    nm, ns
):

Call self as a function.

test_eq(resolve_nm('ca.f', globals()), ca.f)

source

get_schema_nm


def get_schema_nm(
    nm:str, ns, dot2dash:bool=False, kwargs:VAR_KEYWORD
):

Get schema for symbol nm in namespace ns, preserving the full dotted name

schema = get_schema_nm('ca.f', locals())
test_eq(schema['name'], 'ca.f')
schema

{'name': 'ca.f',
 'description': 'Do a thing',
 'input_schema': {'type': 'object',
  'properties': {'a': {'description': 'That is `a`', 'type': 'integer'}},
  'required': ['a']}}

source

call_func


def call_func(
    fc_name, fc_inputs, ns, raise_on_err:bool=True
):

Call the function fc_name with the given fc_inputs using namespace ns.

Now when we an LLM responses with the tool to use and its inputs, we can simply use the same namespace it was given to look up the tool and call it.

call_func('sums', {'a': 1, 'b': 2}, ns=[sums])

assert "unsupported operand type(s) for +: 'int' and 'str'" in call_func('sums', {'a': 1, 'b': '3'}, ns=ns, raise_on_err=False)

test_fail(call_func, args=['sums', {'a': 1, 'b': '3'}], kwargs={'ns': ns})

Types that can be constructed from a plain str can be used directly, as long as they are in custom_types (which you can add to).

def path_test(
    a: Path,  # a type hint
    b: Path   # b type hint
):
    "Mandatory docstring"
    return a/b

test_eq(call_func('path_test', {'a': '/home', 'b': 'user'}, ns=[path_test]), Path('/home/user'))

test_eq(call_func('ca.f', {'a': 5}, ns=globals()), 1)

Async function calling

async def asums(a, b): return a + b
ns = mk_ns(asums)
ns

{'asums': <function __main__.asums(a, b)>}

source

call_func_async


async def call_func_async(
    fc_name, fc_inputs, ns, raise_on_err:bool=True
):

Awaits the function fc_name with the given fc_inputs using namespace ns.

Testing async call_func_async both with sync and async functions.

test_eq(await call_func_async('asums', {'a': 1, 'b': 2}, ns=[asums]), 3)

test_eq(await call_func_async('sums', {'a': 1, 'b': 2}, ns=[sums]), 3)

r = await call_func_async('asums', {'a': 1, 'b': '2'}, ns=[asums], raise_on_err=False)
assert "unsupported operand type(s) for +: 'int' and 'str'" in r

ex = False
try: await call_func_async('asums', {'a': 1, 'b': '2'}, ns=[asums], raise_on_err=True)
except: ex = True
assert ex

class B:
    async def g(self, x:int): return x*2

b = B()
res = await call_func_async('b.g', {'x': 5}, ns=globals())
test_eq(res, 10)

Schema to function

source

mk_param


def mk_param(
    orig, props, req, pynm:NoneType=None
):

Create a Parameter for orig with schema props

tool = dict2obj(dict(
    description='Find real-…',
    inputSchema={
        '$schema': 'http://json-schema.org/draft-07/schema#',
        **dict(
            additionalProperties=False,
            properties=dict(
                language=dict(description='Filter by …', items={'type': 'string'}, type='array'),
                matchCase=dict(default=False, description='Whether th…', type='boolean'),
                path=dict(description='Filter by …', type='string'),
                query=dict(description='The litera…', type='string'),
                useRegexp=dict(default=False, description='Whether to…', type='boolean')),
            required=['query'],
            type='object')},
    name='searchGitHub'))

props, req = tool.inputSchema['properties'], tool.inputSchema['required']
list(props)

['language', 'matchCase', 'path', 'query', 'useRegexp']

props.matchCase

{'default': False, 'description': 'Whether th…', 'type': 'boolean'}

p = mk_param('query', props.query, req)
p, p.kind

(<Parameter "query: str">, <_ParameterKind.POSITIONAL_OR_KEYWORD: 1>)

p = mk_param('language', props.language, req)
p, p.kind

(<Parameter "language: list[str] = None">, <_ParameterKind.KEYWORD_ONLY: 3>)

source

schema2sig


def schema2sig(
    tool
):

Convert json schema tool to a Signature

schema2sig(tool)

<Signature (query: str, *, language: list[str] = None, matchCase: bool = False, path: str = None, useRegexp: bool = False)>

source

mk_tool


def mk_tool(
    dispfn, tool
):

Create a callable function from a JSON schema tool definition

mk_tool is the inverse of get_schema — it creates a callable Python function from a JSON schema tool definition. This is useful for MCP clients where tools are defined as schemas but need to be called as regular Python functions.

The created function has a proper signature, docstring, and annotations, so it works well with IDE autocomplete and introspection.

def dispatch_eg(name, **kwargs): return f"Called {name} with {kwargs}"

fn = mk_tool(dispatch_eg, tool)
fn('hello', path='src/')

"Called searchGitHub with {'query': 'hello', 'path': 'src/'}"

tool_hy = dict2obj(dict(name='run', description='Run command', inputSchema=dict(type='object', properties={'cmd': {'type': 'string'}, 'approval-policy': {'type': 'string', 'default': 'never'}}, required=['cmd'])))

sig = schema2sig(tool_hy)
test_eq(list(sig.parameters), ['cmd', 'approval_policy'])
test_eq(sig.parameters['approval_policy'].default, 'never')
test_eq(sig.parameters['approval_policy'].kind, Parameter.KEYWORD_ONLY)

seen = []
def disp(name, **kwargs):
    seen.append((name, kwargs))
    return kwargs

fn = mk_tool(disp, tool_hy)
test_eq(fn('ls', approval_policy='never'), {'cmd': 'ls', 'approval-policy': 'never'})
test_eq(seen[-1], ('run', {'cmd': 'ls', 'approval-policy': 'never'}))

tool_req_hy = dict2obj(dict(name='run', description='Run command', inputSchema=dict(type='object', properties={'approval-policy': {'type': 'string'}, 'cmd': {'type': 'string'}}, required=['approval-policy', 'cmd'])))

seen = []
fn = mk_tool(disp, tool_req_hy)
fn('never', 'ls')
test_eq(seen[-1], ('run', {'approval-policy': 'never', 'cmd': 'ls'}))

tool_collision = dict2obj(dict(name='run', description='Run command', inputSchema=dict(type='object', properties={'approval-policy': {'type': 'string'}, 'approval_policy': {'type': 'string'}})))

test_fail(lambda: schema2sig(tool_collision), contains='collision')

test_eq(_py_nm(''), '_')

Function calling

Function to schema

get_schema

Usage examples

Additional get_schema() Test Cases

Python tool

python

Tool Calling

mk_ns

resolve_nm

get_schema_nm

call_func

Async function calling

call_func_async

Schema to function

mk_param

schema2sig

mk_tool

Additional `get_schema()` Test Cases