Tools

Tool discovery, schemas and invocation.

registry

Tool resolution and execution across local functions and MCP servers.

Supports local Python modules, MCP stdio subprocesses, and MCP HTTP endpoints.

ToolError

Bases: RuntimeError

Raised when a tool cannot be resolved or fails during execution.

ToolSpec dataclass

ToolSpec(name, provider, description='', parameters=dict())

Specification of a tool callable by agents.

Attributes:

Name	Type	Description
name	str	Tool name.
provider	str	Name of the provider providing the tool.
description	str	Explanation of what the tool does.
parameters	dict[str, Any]	JSON schema dictionary describing input arguments.

qualified property

qualified

Returns the fully qualified name (provider.name).

signature

signature()

Returns a Python-like signature representation for prompting.

Source code in python/src/science_adk/tools/registry.py

def signature(self) -> str:
    """Returns a Python-like signature representation for prompting."""
    props = (self.parameters or {}).get("properties", {}) or {}
    required = set((self.parameters or {}).get("required", []) or [])
    args = []
    for pname, schema in props.items():
        ptype = schema.get("type", "Any")
        hint = {
            "string": "str",
            "integer": "int",
            "number": "float",
            "boolean": "bool",
            "array": "list",
            "object": "dict",
        }.get(ptype, "Any")
        args.append(pname if pname in required else f"{pname}: {hint} = ...")
    return f"{self.name}({', '.join(args)})"

to_dict

to_dict()

Serializes the ToolSpec to a dictionary.

Source code in python/src/science_adk/tools/registry.py

def to_dict(self) -> dict[str, Any]:
    """Serializes the ToolSpec to a dictionary."""
    return {
        "name": self.name,
        "provider": self.provider,
        "description": self.description,
        "parameters": self.parameters,
    }

LocalBackend

LocalBackend(name, path)

Exposes top-level Python functions in a directory as tools.

Source code in python/src/science_adk/tools/registry.py

def __init__(self, name: str, path: Path):
    self.name = name
    self.path = path
    self._functions: dict[str, Callable[..., Any]] = {}
    self._specs: dict[str, ToolSpec] = {}
    self._loaded = False

list_tools

list_tools()

Lists tool specifications provided by this local directory.

Source code in python/src/science_adk/tools/registry.py

def list_tools(self) -> list[ToolSpec]:
    """Lists tool specifications provided by this local directory."""
    self._load()
    return list(self._specs.values())

call

call(name, arguments)

Executes a local tool function.

Source code in python/src/science_adk/tools/registry.py

def call(self, name: str, arguments: dict[str, Any]) -> Any:
    """Executes a local tool function."""
    self._load()
    func = self._functions.get(name)
    if func is None:
        raise ToolError(f"Local tool {name!r} not found in {self.path}")
    return func(**arguments)

McpStdioBackend

McpStdioBackend(name, command, env=None)

Communicates with an MCP server subprocess via stdio JSON-RPC.

Source code in python/src/science_adk/tools/registry.py

def __init__(self, name: str, command: str, env: dict[str, str] | None = None):
    self.name = name
    self.command = command
    self.env = env or {}
    self._process: subprocess.Popen | None = None
    self._counter = 0
    self._lock = threading.Lock()

list_tools

list_tools()

Lists tools exposed by the stdio MCP server.

Source code in python/src/science_adk/tools/registry.py

def list_tools(self) -> list[ToolSpec]:
    """Lists tools exposed by the stdio MCP server."""
    self._start()
    result = self._request("tools/list", {})
    return [
        ToolSpec(
            name=tool["name"],
            provider=self.name,
            description=tool.get("description", ""),
            parameters=tool.get("inputSchema", {}),
        )
        for tool in result.get("tools", [])
    ]

call

call(name, arguments)

Executes a tool on the stdio MCP server.

Source code in python/src/science_adk/tools/registry.py

def call(self, name: str, arguments: dict[str, Any]) -> Any:
    """Executes a tool on the stdio MCP server."""
    self._start()
    result = self._request("tools/call", {"name": name, "arguments": arguments})
    return _unwrap_mcp_result(result)

close

close()

Terminates the subprocess.

Source code in python/src/science_adk/tools/registry.py

def close(self) -> None:
    """Terminates the subprocess."""
    if self._process and self._process.poll() is None:
        self._process.terminate()
        try:
            self._process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self._process.kill()

McpHttpBackend

McpHttpBackend(name, url, token='', timeout=120)

Communicates with a remote MCP server over HTTP.

Source code in python/src/science_adk/tools/registry.py

def __init__(self, name: str, url: str, token: str = "", timeout: int = 120):
    self.name = name
    self.url = url.rstrip("/")
    self.token = token
    self.timeout = timeout
    self._counter = 0

list_tools

list_tools()

Lists tools exposed by the remote HTTP MCP server.

Source code in python/src/science_adk/tools/registry.py

def list_tools(self) -> list[ToolSpec]:
    """Lists tools exposed by the remote HTTP MCP server."""
    result = self._post("tools/list", {})
    return [
        ToolSpec(
            name=tool["name"],
            provider=self.name,
            description=tool.get("description", ""),
            parameters=tool.get("inputSchema", {}),
        )
        for tool in result.get("tools", [])
    ]

call

call(name, arguments)

Executes a tool on the remote HTTP MCP server.

Source code in python/src/science_adk/tools/registry.py

def call(self, name: str, arguments: dict[str, Any]) -> Any:
    """Executes a tool on the remote HTTP MCP server."""
    return _unwrap_mcp_result(
        self._post("tools/call", {"name": name, "arguments": arguments})
    )

close

close()

No-op for HTTP backend.

Source code in python/src/science_adk/tools/registry.py

def close(self) -> None:
    """No-op for HTTP backend."""
    return None

ToolRegistry

ToolRegistry(providers, root)

Aggregates and resolves tools across all configured providers.

Source code in python/src/science_adk/tools/registry.py

def __init__(self, providers: list[ToolProvider], root: Path):
    self.root = Path(root)
    self._backends: dict[str, Any] = {}
    self._index: dict[str, ToolSpec] | None = None
    for provider in providers:
        if provider.enabled:
            self._backends[provider.name] = self._build(provider)

from_project classmethod

from_project(project)

Constructs a ToolRegistry from a Project instance.

Source code in python/src/science_adk/tools/registry.py

@classmethod
def from_project(cls, project) -> ToolRegistry:
    """Constructs a ToolRegistry from a Project instance."""
    return cls(project.config().providers, project.root)

specs

specs(refresh=False)

Returns all ToolSpecs from all active providers.

Source code in python/src/science_adk/tools/registry.py

def specs(self, refresh: bool = False) -> list[ToolSpec]:
    """Returns all ToolSpecs from all active providers."""
    if self._index is None or refresh:
        index: dict[str, ToolSpec] = {}
        for name, backend in self._backends.items():
            try:
                for spec in backend.list_tools():
                    index[spec.qualified] = spec
            except ToolError as exc:
                print(
                    f"warning: tool provider {name!r} unavailable: {exc}",
                    file=sys.stderr,
                )
        self._index = index
    return list(self._index.values())

resolve

resolve(name)

Resolves a tool by qualified or unqualified name.

Parameters:

Name	Type	Description	Default
name	str	Qualified ('provider.tool') or bare tool name.	required

Returns:

Type	Description
ToolSpec	Matching ToolSpec.

Raises:

Type	Description
ToolError	If tool is missing or name is ambiguous.

Source code in python/src/science_adk/tools/registry.py

def resolve(self, name: str) -> ToolSpec:
    """Resolves a tool by qualified or unqualified name.

    Args:
      name: Qualified ('provider.tool') or bare tool name.

    Returns:
      Matching ToolSpec.

    Raises:
      ToolError: If tool is missing or name is ambiguous.
    """
    specs = self.specs()
    if "." in name:
        for spec in specs:
            if spec.qualified == name:
                return spec
    matches = [s for s in specs if s.name == name]
    if len(matches) == 1:
        return matches[0]
    if len(matches) > 1:
        options = ", ".join(sorted(m.qualified for m in matches))
        raise ToolError(
            f"Tool name {name!r} is ambiguous. Qualify it as one of: {options}"
        )
    available = sorted(s.name for s in specs)
    hint = ", ".join(available[:12]) or "none configured"
    raise ToolError(f"Unknown tool {name!r}. Available: {hint}")

call

call(name, arguments)

Invokes a tool by name with arguments.

Source code in python/src/science_adk/tools/registry.py

def call(self, name: str, arguments: dict[str, Any]) -> Any:
    """Invokes a tool by name with arguments."""
    spec = self.resolve(name)
    backend = self._backends[spec.provider]
    return backend.call(spec.name, arguments)

close

close()

Closes all active provider backends.

Source code in python/src/science_adk/tools/registry.py

def close(self) -> None:
    """Closes all active provider backends."""
    for backend in self._backends.values():
        close_func = getattr(backend, "close", None)
        if close_func:
            try:
                close_func()
            except Exception:
                pass