Skip to main content

Tools

Tools extend agent capabilities with additional functions. They enable LLMs and agents to interact with external systems, execute code, access data and perform specialized tasks that go beyond pure text generation.

ProtoLink derives argument schemas from type hints, validates each call and applies the agent's policies before execution. The model can select a registered tool, or your application can call a known tool directly without a model.

from protolink import Agent


def add(a: int, b: int) -> int:
"""Add two integers."""
return a + b


agent = Agent(name="calculator", tools=[add])
print(agent.sync.call_tool("add", a=2, b=3)) # 5

This example runs offline. Use await agent.call_tool(...) in async code, or attach an LLM and use invoke() when the model should decide which tool to call.

Use Agent(sync_execution="thread", max_sync_workers=4) for blocking synchronous functions. @agent.tool(sync_execution="inline") and Tool.from_callable(func, sync_execution="thread") override that default per tool. Async tools are awaited directly. Worker calls preserve context variables and native runtime controls; cancellation cannot terminate a running Python thread. See application execution.

Register a collection or MCP server​

agent.add_tools(iterable) applies ordinary add_tool registration to tools and typed functions. await agent.add_mcp(command=..., args=..., include=[...], prefix="...") discovers and registers selected stdio tools; pass url=... for SSE or a configured MCPToolAdapter for full control. Blocking scripts use agent.sync.add_mcp(...). The adapter also exposes list_tools_async and get_tools_async. See collection and MCP registration for collision handling, discovery side effects and connection ownership.

See Built-in Tools for the complete built-in catalog. Execution, approvals and recovery covers run handles, approval brokers and completion checks.

Reject a request the model can correct​

Schema validation checks the shape of a request. A tool often also needs application validation: an ID must exist, a selection must belong to the current catalog, or a requested combination must be supported. Raise ToolRejectedError when this validation rejects a request before any changes or external effects occur. The inference loop returns the structured feedback to the same model, which can correct its next request, choose a different action, or explain why it cannot proceed.

from protolink import Agent, ToolRejectedError, create_llm

checks = {"lint": "Check source style", "tests": "Run the test suite"}
plan = []


def plan_task(check_id: str) -> dict[str, str]:
"""Select a known check before changing the plan."""
if check_id not in checks:
raise ToolRejectedError(
"Unknown check selection.",
code="invalid_check_selection",
details={"valid_ids": list(checks)},
suggestion="Select one of the advertised check IDs.",
)
plan.append(check_id)
return {"check_id": check_id, "description": checks[check_id]}


llm = create_llm(
"ollama:qwen3:4b",
base_url="http://localhost:11434",
tool_calling="auto",
max_tool_corrections=2,
)
agent = Agent(name="planner", llm=llm, tools=[plan_task])
answer = agent.sync.invoke("Select a suitable check.")

Only the tool's accepted branch changes plan. The model receives the error code, message, valid IDs and suggestion, alongside status="rejected" and changes_made=false. JSON and native action acquisition use the same correction machinery and the adapter's existing tool-observation hook. Try the runnable offline example without a model server.

Bound corrections within the current inference​

max_tool_corrections defaults to two corrective turns per infer() invocation. Local and delegated tool validation, unavailable delegation routes and explicit no-change rejections share this allowance. Two rejected requests can receive corrective turns; a third rejection raises ToolCorrectionLimitError. Set the limit to 0 to stop at the first rejection, or assign llm.max_tool_corrections after construction. The accepted range is 0 through 10; the inference loop's existing ten-step ceiling can stop work sooner.

Successful actions do not replenish the allowance. Every subsequent model call, step and actual tool invocation still consumes the existing run budgets; an invoked tool that rejects counts as a tool call. A schema or unknown-tool rejection never invokes the callable and does not charge a tool call. Completed observations remain available and successful actions retain their deduplication protection. Rejected requests are not marked as successful, so the model can correct and resubmit them. No automatic tool retry occurs outside a model-driven inference loop.

An action_builder or PreparedTool.prepare callback can also raise an explicit ToolRejectedError before approval or execution. The same correction allowance applies; feedback records phase="preparation" and consumes no tool call because the executor never ran. Preparation must make no changes when it rejects. Ordinary builder exceptions, policy denial and uncertain execution still stop the run. Cancellation and hard budgets take precedence over correction. See the prepared proposal walkthrough.

ToolCorrectionLimitError exposes tool_name, attempts, limit and last_error. A limit failure concerns the latest rejected request; earlier successful actions may already have changed resources. Agent task failures include a tool_correction_limit blocker with the final rejection and counters.

Distinguish rejection from execution failure​

OutcomeContractReporting
Successful executionThe tool returned its resulttool_result / action.completed
Request rejectedToolRejectedError, or pre-execution tool/schema validation; nothing changed in this requesttool_rejected / action.rejected
Delegation rejectedUnsupported advertised target/tool/schema, or a correlated single-tool no-change responseagent_call_rejected / action.rejected
Permission deniedPolicy or approval denied the actionaction.denied
Execution failedOrdinary exception inside the tool; effects may be partial or unknowntool_error / action.failed

A rejection is an application guarantee, not a rollback performed by ProtoLink. Do not convert arbitrary exceptions, network failures, partial writes or uncertain external responses into ToolRejectedError. Keep them as execution failures and use your application's reconciliation procedure where necessary. Raising ValueError inside a tool continues to fail inference; it is not treated as argument-schema feedback.

Keep rejection details concise and safe to expose: the model, telemetry, client event streams and persisted reports can see this feedback. Details must be a JSON object with serializable, finite values. after_tool hooks receive successful observations only and do not run for rejections. Explicit tool objects and agent.call_tool() propagate the rejection without correction; task-level tool execution returns a structured tool_output.error. Durable direct calls use that task contract and can raise TaskExecutionError.

With durability enabled, ProtoLink commits a distinct rejection receipt and preserves the correction counter. Approval pauses and application restarts do not reset the allowance. If the process stops after the receipt is committed but before feedback enters history, resumption reuses the recorded rejection without calling or charging the tool again. An action_rejection artifact never proves successful execution for CompletionValidator.

Delegated tool calls use the same validation principles. The sender checks advertised tool names and schemas before dispatch; native receivers return structured rejection errors for unknown tools or invalid arguments before entering user code. A single-tool no-change response can be corrected by the parent's model. Ordinary remote failures and completed effects remain terminal. See delegation capability checks.

ToolRejectedError​

exceptionprotolink.ToolRejectedError
source
ToolRejectedError(
  message: str,
  *,
  code: str = "tool_rejected",
  details: dict[str, Any] | None = None,
  suggestion: str | None = None,
)

Available from protolink and protolink.tools. Raise only for a rejected request that made no changes. message and code must be nonblank strings; suggestion is nonblank text or None. details is copied through JSON serialization. Invalid feedback raises TypeError or ValueError and is not silently recovered. to_dict() returns detached feedback with the no-change outcome; from_dict() restores a committed rejection.

ToolCorrectionLimitError​

exceptionprotolink.ToolCorrectionLimitError
source
ToolCorrectionLimitError(
  *,
  tool_name: str,
  attempts: int,
  limit: int,
  last_error: ToolRejectedError,
)

Terminal inference exception available from protolink and protolink.tools. Applications normally inspect this exception rather than construct it. attempts counts rejected requests; limit counts permitted corrective turns. The final rejection is available through last_error, including its code, details and suggestion.

Overview​

Protolink provides a flexible tool system with three approaches:

  • Built-in Tools: Opt-in factories for general capabilities and configured integrations; see Built-in Tools
  • Native Tools: Python functions decorated directly on an agent
  • MCP Tools: Tools from external MCP (Model Context Protocol) servers

All three tool sources use the same interface, making them interchangeable from the agent's perspective.

Module Structure​

The tools module is organized as follows:

# Core interfaces and opt-in built-ins
from protolink.tools import (
BaseTool,
Tool,
calculator,
current_datetime,
fetch_url,
web_search,
)

# Tool adapters for external integrations
from protolink.tools.adapters import MCPToolAdapter
ModuleDescription
protolink.toolsCore interfaces, native implementation and public built-in factories
protolink.tools.builtinsBuilt-in factories and configurable backends
protolink.tools.adaptersAdapters for integrating external tool systems

Tool moduleTools

The callable capability layer for native Python functions, MCP-backed tools, JSON schemas, examples, capability policies and approval-aware action metadata.

protolink.tools
Native decoratorsMCP adaptersSchema inferenceCapability enforcementAgentSkill advertising
Built-in toolsRegister general-purpose capabilities and configured integrations from the Built-in Tools catalog.agent.add_tool(web_search())
Base contractTools are async callables with a name, description, input schema, output schema, tags, examples and capabilities.BaseTool
Native toolsRegister typed Python functions directly on an agent and let Protolink infer schemas.@agent.tool
MCP toolsAdapt tools from Model Context Protocol servers into the same runtime interface.MCPToolAdapter
Policy hooksAttach capabilities and action builders so runtime policy can allow, deny, or request approval.capabilities

BaseTool Protocol​

All tools in ProtoLink conform structurally to BaseTool. It is a typing protocol rather than a concrete base class: any object with the advertised metadata and asynchronous call behavior can be registered, including native Tool instances and wrapped MCP tools.

BaseTool​

protocolprotolink.tools.BaseTool
source
class BaseTool(Protocol):
  name: str
  description: str
  input_schema: dict[str, Any] | None
  output_schema: Any | None
  tags: list[str] | None
  examples: list[Any] | None
  capabilities: Collection[str] | None

  async __call__(**kwargs) -> Any

The protocol is the smallest contract understood by Agent registration and execution. Metadata describes the tool to models, discovery clients and policy; __call__() performs the actual operation.

Attributes

namestr

Stable identifier used in model tool declarations, task parts, registry skills, policy actions and agent.call_tool(). Names should be unique within one Agent because registering the same name replaces the runtime tool.

descriptionstr

Human-readable purpose shown to the model and copied to the advertised AgentSkill. Explain when to call the tool, not only what its Python function is named.

input_schemadict[str, Any] | None

JSON Schema for accepted keyword arguments. Agent execution validates model-provided arguments against this schema before the callable runs.

output_schemaAny | None

Optional schema describing the returned value. It is advertised to callers but does not currently validate the runtime result.

tagslist[str] | None

Discovery and presentation labels copied to the Agent's skill card.

exampleslist[Any] | None

Representative calls or values copied to AgentSkill.examples. They guide clients and models but are not executed automatically.

capabilitiesCollection[str] | None

Authority strings merged into the RunAction evaluated immediately before execution. Capabilities become enforceable only when the call passes through Agent policy.

Call

**kwargsAny

Keyword arguments matching input_schema. Positional invocation is outside the protocol.

returnAny

Tool-specific result. Implementations may return any JSON-compatible or application value.

Runtime boundary

Calling a tool object directly bypasses Agent authorization and approvals. Register it and use agent.call_tool() or agent.sync.call_tool() for those controls. Use task execution for the full cancellation, telemetry, budget and persistence lifecycle.

Tool​

dataclassprotolink.tools.Tool
source
class Tool(
  name: str,
  description: str,
  input_schema: dict[str, Any] | None,
  output_schema: Any | None,
  tags: list[str] | None,
  func: Callable[..., Any],
  args: dict[str, Any] | None = None,
  examples: list[Any] | None = None,
  capabilities: Collection[str] | None = None,
  action_builder: ActionBuilder | None = None,
  sync_execution: Literal["inline", "thread"] | None = None,
)

Adapt a synchronous or asynchronous Python callable to the BaseTool contract. Construction inspects the callable signature and resolved type hints, infers any missing schemas, normalizes explicit schemas and converts missing tags, examples and capabilities into empty collections.

For ordinary functions, use agent.add_tool(func) or @agent.tool to infer name, description and schemas. Use Tool.from_callable(func, ...) for a reusable definition with explicit metadata overrides and the constructor when restoring or assembling a complete tool definition.

Parameters

namestrrequired

Stable runtime and advertised identifier.

descriptionstrrequired

Purpose presented to models and discovery clients.

input_schemadict[str, Any] | Nonerequired

Explicit JSON Schema or legacy field map. Pass None to infer an object schema from func's parameters and type annotations.

output_schemaAny | Nonerequired

Explicit result schema, or None to infer one from the callable's return annotation.

tagslist[str] | Nonerequired

Discovery labels. None becomes an empty list.

funcCallable[..., Any]required

Wrapped Python function. Synchronous return values are accepted; awaitable results are awaited automatically.

argsdict[str, Any] | Nonedefault: None

Legacy metadata retained on the dataclass. Runtime invocation uses arguments passed to call(), not this mapping.

exampleslist[Any] | Nonedefault: None

Advertised examples. None becomes an empty list.

capabilitiesCollection[str] | Nonedefault: None

Required policy capabilities. Empty values are removed and duplicates are discarded while preserving first occurrence.

action_builderActionBuilder | Nonedefault: None

Optional sync or async callback that receives validated arguments and the active RunContext, then returns a customized RunAction with preview artifacts or metadata.

sync_executionLiteral["inline", "thread"] | Nonedefault: None

Override synchronous execution for this tool; None inherits the executing agent's default. Standalone tools default to inline execution. Async callables are awaited directly in either mode.

Raises

ValueError | TypeError

Callable inspection, type-hint resolution, or explicit schema normalization can fail during construction.

Tool.from_callable​

classmethodprotolink.tools.Tool.from_callable
source
Tool.from_callable(
  func: Callable[..., Any],
  *,
  name: str | None = None,
  description: str | None = None,
  input_schema: dict[str, Any] | None = None,
  output_schema: Any | None = None,
  tags: list[str] | None = None,
  examples: list[Any] | None = None,
  capabilities: Collection[str] | None = None,
  action_builder: ActionBuilder | None = None,
  sync_execution: Literal["inline", "thread"] | None = None,
) -> Tool

Create a reusable tool from a synchronous or asynchronous callable. The default name is the function name, or the class name for a callable object. The default description is its cleaned docstring, falling back to Call <name>.. Explicit metadata takes precedence; missing input and output schemas use the same type-hint inference as the constructor. agent.add_tool(func) calls this factory automatically for ordinary callables; use it explicitly when adding metadata overrides or sharing a preconfigured tool.

Parameters

funcCallable[..., Any]required
Original callable retained as tool.func. The factory does not replace or execute it.
name, descriptionstr | Nonedefault: None
Optional public metadata overrides. All arguments after func are keyword-only.
input_schema, output_schema, tags, examples, capabilities, action_builder, sync_execution
The same schema, discovery, permission, preview and execution metadata accepted by the explicit constructor.

Returns

toolTool
A tool ready for agent.add_tool(), with inferred schemas and unchanged validation behavior.

Raises

TypeError
The input is not callable, its signature cannot be inspected, or its explicit or inferred name is not a string.
ValueError
The tool name is empty or whitespace-only, or signature inspection or explicit schema normalization fails.
from protolink import Tool


def add(a: int, b: int) -> int:
"""Add two integers."""
return a + b


add_tool = Tool.from_callable(add, tags=["math"])
agent.add_tool(add_tool)
# The same definition can also be registered on another agent.

Tool.validate_args​

methodprotolink.tools.Tool.validate_args
source
validate_args(
  kwargs: dict[str, Any] | None,
) -> dict[str, Any]

Validate proposed keyword arguments with the normalized input schema, wrapped callable signature and resolved type hints. Custom execution paths can call this method to perform the same coercion as Tool.__call__() before preparing policy metadata.

Parameters

kwargsdict[str, Any] | Nonerequired

Untrusted argument mapping. None becomes an empty dictionary and the supplied mapping is copied before coercion.

Returns

argumentsdict[str, Any]

Validated mapping containing any safe scalar conversions or reconstructed annotated objects.

Raises

ValueError

Schema violations, missing or unexpected fields, incompatible values and annotation-validation failures.

Tool.call​

async methodprotolink.tools.Tool.__call__
source
async __call__(
  **kwargs: Any,
) -> Any

Validate keyword arguments and invoke the wrapped Python callable. A synchronous result is returned from the coroutine immediately; an awaitable result is awaited before returning.

Parameters

**kwargsAny

Values accepted by the generated or explicit input schema and the callable signature.

Returns

resultAny

Direct or awaited result from func. output_schema advertises this value but is not enforced here.

Raises

ValueError

Argument validation fails before user code executes.

callable error

Exceptions from the wrapped function propagate unchanged.

Direct invocation

This method does not apply Agent policy or task controls by itself. Those surround the call in Agent.call_tool_in_context() and the task engine.

Tool.prepare_action​

async methodprotolink.tools.Tool.prepare_action
source
async prepare_action(
  arguments: dict[str, Any],
  context: RunContext,
) -> RunAction

Build the runtime action evaluated before this tool executes. Without a custom builder, the action has kind tool.call, the tool name and description, validated arguments as payload and the tool's declared capability set.

Parameters

argumentsdict[str, Any]required

Already validated keyword arguments proposed for execution.

contextRunContextrequired

Active run identity, permissions, session, cancellation and budget context supplied to a custom builder.

Returns

actionRunAction

Domain-neutral policy action. A sync or async custom builder may add metadata or preview artifacts, but the tool's required capabilities are merged back into the result.

Raises

TypeError

The configured action_builder does not return RunAction.

builder error

Exceptions raised by the application builder propagate to the execution layer.

ToolRejectedError

Explicit no-change preparation rejection. Model-driven inference returns bounded correction feedback before approval or execution; direct tool tasks retain their structured rejection behavior.

All tools are async callables from the caller's perspective and accept keyword arguments matching their input schema:

# Tools are invoked with keyword arguments
result = await tool(location="Tokyo", units="celsius")

Built-in Tools​

ProtoLink provides optional web search/fetch, calculator/clock, process/shell/Git, user feedback, filesystem, storage, HTTP, document, database, calendar and email tools. Register the factories on any Agent and supply the resources, backends and policies your application needs.

See Built-in Tools for the complete catalog, generated calls, configuration, authentication, limits and examples. The Built-in Agents page documents the small agent presets that compose these capabilities.

Native Tools​

Native tools are regular Python callables that you register on an agent. They are exposed over the transport so that other agents (or clients) can invoke them.

Registering Native Tools​

Pass an existing function to agent.add_tool() or decorate a function with @agent.tool. Both infer the public name, cleaned docstring and schemas from the callable:

from protolink import Agent, AgentCard

agent_card = AgentCard(url="runtime://calculator", name="calculator_agent", description="Agent with math tools")
agent = Agent(card=agent_card, transport="runtime", verbosity=0)


def add(a: int, b: int) -> int:
"""Add two integers and return the result."""
return a + b


agent.add_tool(add)


@agent.tool()
async def multiply(a: float, b: float) -> float:
"""Multiply two numbers and return the result."""
return a * b


print(agent.sync.call_tool("add", a=2, b=3)) # 5

# Inferred Schemas:
# input_schema: {
# "type": "object",
# "properties": {
# "a": {"type": "integer"},
# "b": {"type": "integer"}
# },
# "required": ["a", "b"],
# "additionalProperties": False
# }
# output_schema: {"type": "integer"}

Synchronous functions, asynchronous functions, bound methods and callable objects can all be passed directly to add_tool(). To reuse add on another agent, call other_agent.add_tool(add). Registration inspects the callable without executing it; the registered wrapper is available as agent.tools["add"].

Existing Tool, built-in, MCP and custom tool objects retain their metadata and behavior when passed to add_tool(). For explicit schemas, tags, capabilities, or approval previews, create a configured Tool.from_callable(add, ...) first. See Agent.add_tool for replacement rules.

Agent.tool​

methodprotolink.agents.Agent.tool
source
tool(
  name: str | ToolCallableT | None = None,
  description: str | None = None,
  input_schema: dict[str, Any] | None = None,
  output_schema: dict[str, Any] | None = None,
  tags: list[str] | None = None,
  examples: list[Any] | None = None,
  capabilities: list[str] | tuple[str, ...] | set[str] | None = None,
  action_builder: ActionBuilder | None = None,
) -> ToolCallableT | Callable[[ToolCallableT], ToolCallableT]

Wrap a Python callable in Tool, register it immediately on this Agent and synchronize the corresponding advertised AgentSkill. The decorated name remains bound to the original function and retains its type signature, while the runtime wrapper is available through agent.tools[name]. ToolCallableT represents the decorated callable's type. Bare decoration, empty parentheses, explicit keywords and the existing @agent.tool("name", "description") form are supported.

Parameters

namestr | ToolCallableT | Nonedefault: None

Stable public identifier, inferred from the function name when omitted. Bare decoration passes the callable here. Reusing an existing runtime name replaces the tool and its matching skill.

descriptionstr | Nonedefault: None

Selection guidance shown to the LLM and discovery clients. Defaults to the cleaned docstring, or Call <name>. when the function has none. Include the operation's purpose, prerequisites and relevant side effects in your docstring or explicit description.

input_schemadict[str, Any] | Nonedefault: None

Explicit JSON Schema or legacy field map. None infers a schema from the decorated function's signature and type hints.

output_schemadict[str, Any] | Nonedefault: None

Explicit result schema. None infers it from the return annotation; it is advertised but does not validate the returned runtime value.

tagslist[str] | Nonedefault: None

Discovery labels copied to the generated AgentSkill.

exampleslist[Any] | Nonedefault: None

Representative examples copied to the skill card.

capabilitieslist[str] | tuple[str, ...] | set[str] | Nonedefault: None

Authority required before Agent execution. The wrapper normalizes non-empty values and merges them into every prepared action.

action_builderActionBuilder | Nonedefault: None

Optional sync or async builder for action metadata and approval-preview artifacts. It runs after argument validation and before policy evaluation.

Returns

function or decoratorToolCallableT | Callable[[ToolCallableT], ToolCallableT]

Bare decoration returns the original function; the configured form returns a decorator that registers and returns it unchanged. Direct Python calls bypass Agent authorization and approvals.

Registration timing

Registration occurs when Python evaluates the decorated function definition, not when the tool is first called. In skills="auto" mode the Agent card is updated at the same time.

JSON Schema and Runtime Validation​

Tool schemas are first-class JSON Schema objects. Native tools infer nested schemas from Python type hints, dataclasses, enums, typed dictionaries and Pydantic models. Before execution, Protolink validates and lightly coerces tool arguments against the schema, then applies Python annotation validation where available.

from pydantic import BaseModel, Field


class BookingRequest(BaseModel):
location: str
guests: int = Field(gt=0)


@agent.tool(
name="book_hotel",
description="Book a hotel",
examples=[{"booking": {"location": "Athens", "guests": 2}}],
)
async def book_hotel(booking: BookingRequest) -> dict[str, str]:
return {"location": booking.location, "status": "confirmed"}

The inferred input schema is a JSON Schema object with a nested booking property. Runtime calls such as {"booking": {"location": "Athens", "guests": "2"}} are coerced before the function receives a BookingRequest instance. Missing required fields, unexpected fields, invalid enums and incompatible scalar values fail validation before reaching user code. Inference reports these as no-change rejections; direct Tool validation raises ValueError.

Schema helper API​

The helpers below are public for applications that build custom tool wrappers or want to inspect exactly what Tool will infer.

normalize_schema​

functionprotolink.tools.normalize_schema
source
normalize_schema(
  schema: Any,
  title: str | None = None,
) -> dict[str, Any]

Normalize a full JSON Schema, a Pydantic model, a Python annotation, or a legacy {field: type} map into one JSON Schema dictionary. Object schemas receive stable defaults for properties, required and additionalProperties and local $ref definitions are inlined for provider portability.

Parameters

schemaAnyrequired

Supported schema representation. None becomes an empty closed object schema; dictionaries that already look like JSON Schema are copied before normalization.

titlestr | Nonedefault: None

Optional title written onto the returned top-level schema.

Returns

schemadict[str, Any]

New normalized schema dictionary. The input dictionary is not mutated.

infer_input_schema​

functionprotolink.tools.infer_input_schema
source
infer_input_schema(
  func: Callable[..., Any],
  *,
  title: str,
) -> dict[str, Any]

Inspect a callable and build a closed object schema for its named parameters. self, cls, *args and **kwargs are omitted; parameters without Python defaults become required and parameters with defaults include that value in their property schema.

Parameters

funcCallable[..., Any]required

Function whose signature and resolved type hints describe tool input.

titlestrrequired

Required schema title, normally derived from the tool name.

Returns

schemadict[str, Any]

JSON Schema object with additionalProperties=False.

infer_output_schema​

functionprotolink.tools.infer_output_schema
source
infer_output_schema(
  func: Callable[..., Any],
  *,
  title: str,
) -> dict[str, Any]

Convert a callable's resolved return annotation to JSON Schema. Missing or unresolvable annotations produce a permissive schema rather than inspecting or executing the function.

Parameters

funcCallable[..., Any]required

Callable whose return type should be advertised.

titlestrrequired

Title added to the returned schema.

Returns

schemadict[str, Any]

JSON Schema describing the annotated return value.

validate_tool_args​

functionprotolink.tools.validate_tool_args
source
validate_tool_args(
  args: dict[str, Any] | None,
  input_schema: dict[str, Any] | None,
  *,
  type_hints: dict[str, Any] | None = None,
  signature: inspect.Signature | None = None,
) -> dict[str, Any]

Validate and coerce untrusted keyword arguments before tool code runs. JSON Schema validation happens first, Python signature checks catch missing and unexpected fields next and resolved annotations can finally reconstruct typed values through Pydantic TypeAdapter.

Parameters

argsdict[str, Any] | Nonerequired

Proposed arguments. None is normalized to an empty dictionary and the caller's mapping is copied.

input_schemadict[str, Any] | Nonerequired

Schema used for structural validation and conservative coercion of strings, numbers, booleans, arrays and nested objects.

type_hintsdict[str, Any] | Nonedefault: None

Resolved annotations keyed by parameter name. When supplied, matching values are validated and reconstructed after schema checks.

signatureinspect.Signature | Nonedefault: None

Callable signature used to detect required and unexpected keyword fields. A callable accepting **kwargs permits additional names.

Returns

argumentsdict[str, Any]

New validated mapping, potentially containing coerced scalars or reconstructed annotated objects.

Raises

ValueError

Schema violations, missing fields, unexpected fields, incompatible scalar values, invalid enums or constants and annotation-validation failures.

Capabilities And Approval​

Declare capabilities for operations that should participate in runtime policy. Capability names are extensible strings rather than a fixed coding or filesystem taxonomy.

from protolink import Agent, ApprovalDecision, CapabilityPolicy


async def approve(request, context):
return ApprovalDecision(approved=True, request_id=request.request_id)


agent = Agent(
card,
policy=CapabilityPolicy({"records.write": "require_approval"}),
approval_handler=approve,
)


@agent.tool(
name="publish_record",
description="Publish one record",
capabilities=["records.write"],
)
async def publish_record(record_id: str) -> dict[str, str]:
return {"record_id": record_id, "status": "published"}

Policy is evaluated after argument validation and immediately before the callable runs. See Runtime for wildcard rules, RunContext.permissions, approval handlers and preview artifacts.

Use agent.call_tool_in_context(name, context, **arguments) to supply explicit per-run permissions and approval context. agent.call_tool() and agent.sync.call_tool() create a fresh context and return the raw result. These direct calls do not register a task or create its lifecycle, telemetry and persistence records; use run_task(Task.create_tool_call(...)) when the full task boundary is needed.

Tool Cancellation​

Tools invoked by a running task participate in that task's live cancellation automatically. Protolink checks the token before authorization, before calling the tool and after the awaited result returns. It also cancels the owning task, so an async tool normally receives asyncio.CancelledError at its current await point.

@agent.tool(name="build_report", description="Build a report in stages")
async def build_report() -> str:
data = await load_data()
report = await render_report(data)
await commit_report(report)
return "committed"

Cancellation can interrupt the first two awaits, but it cannot undo a commit that an external system already accepted. Side-effecting tools should therefore delay irreversible commits, use transactional APIs, or forward cancellation to a subprocess or remote service that supports it.

Synchronous Python tools cannot be forcibly stopped safely. If they run on the event-loop thread, the cancellation request is processed only after they return. If an application moves them to a worker thread, the event loop remains responsive but the thread itself may continue. These limits are why Protolink defines cancellation as best-effort rather than a rollback guarantee.

When to Use Native Tools​

Native tools are ideal for:

  • Business logic: Domain-specific operations like order processing, data validation
  • Data access: Database queries, API calls, file operations
  • Computation: Complex calculations, data transformations
  • System integration: Interacting with internal services

Tool Tags​

Tools can be categorized using tags for better organization and discovery:

@agent.tool(name="calculate", description="Performs arithmetic calculations", tags=["math", "utility"])
async def calculate(operation: str, a: float, b: float) -> float:
"""Perform basic arithmetic operations."""
if operation == "add":
return a + b
elif operation == "subtract":
return a - b
elif operation == "multiply":
return a * b
elif operation == "divide":
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
else:
raise ValueError(f"Unsupported operation: {operation}")


@agent.tool(name="search_documents", description="Search internal documents", tags=["search", "documents", "rag"])
async def search_documents(query: str, limit: int = 10) -> list[dict]:
"""Search the document database."""
# Implementation here
pass

Tags are automatically propagated to the agent's skills and can be used for:

  • Filtering: Find tools by category
  • Discovery: Help users understand available capabilities
  • Organization: Group related tools together

MCP Tools​

Protolink integrates seamlessly with MCP (Model Context Protocol) servers, allowing you to use tools from external MCP-compatible services as if they were native tools.

What is MCP?​

The Model Context Protocol is an open standard for connecting AI assistants to external tools and data sources. MCP servers can be:

  • Local Python scripts running as subprocesses
  • Remote web services exposing Streamable HTTP or legacy SSE endpoints
  • Third-party tool providers

MCPToolAdapter​

The MCPToolAdapter class connects to MCP servers and exposes their tools as callables compatible with Protolink's BaseTool protocol.

Supported Transports​

TransportDescriptionUse Case
stdioLocal subprocess via stdin/stdoutLocal Python/Node.js MCP servers
streamable_httpStreamable HTTPRemote MCP web services
sseLegacy Server-Sent Events over HTTPOlder MCP web services

Constructor​

classprotolink.tools.adapters.MCPToolAdapter
source
class MCPToolAdapter(
  transport: str = "stdio",
  *,
  command: str | None = None,
  args: list[str] | None = None,
  url: str | None = None,
  headers: dict[str, str] | None = None,
)

Store the connection configuration for an MCP server and provide discovery and wrapping helpers. Construction does not open a subprocess, network connection, or MCP session. Each operation initializes its own session unless a managed adapter.session() context is active.

Parameters

transportstrdefault: "stdio"

MCP client transport: "stdio", "streamable_http", or legacy "sse".

commandstr | Nonedefault: None

Executable launched for stdio, such as "python", "node", or an MCP server binary. It is required when the first stdio operation runs.

argslist[str] | Nonedefault: None

Arguments passed unchanged to the stdio command. None becomes an empty list.

urlstr | Nonedefault: None

Endpoint URL required for either HTTP transport.

headersdict[str, str] | Nonedefault: None

Headers forwarded by the selected HTTP client, commonly for authentication. None becomes an empty dictionary.

Raises

ImportError

Importing protolink.tools.adapters fails when the optional MCP dependency is unavailable. Install protolink[mcp].

ValueError

Discovery or invocation raises for an unknown transport, missing stdio command, or missing HTTP URL. Configuration is validated lazily, not by the constructor.

Session lifetime

By default, each uncached discovery or tool invocation opens and closes its own session. Use async with adapter.session() to share one connection across discovery and registered tools. Await all calls before leaving the context; open and close it in the same task. Nested contexts and calls from another event loop are rejected. Tool calls are never automatically retried.


Connecting to MCP Servers​

Streamable HTTP and connection reuse​

from protolink.tools.adapters import MCPToolAdapter

adapter = MCPToolAdapter(
"streamable_http",
url="https://example.com/mcp",
headers={"Authorization": "Bearer token"},
)
async with adapter.session():
await agent.add_mcp(adapter, include=["search"], prefix="remote_")
result = await agent.call_tool("remote_search", query="agent protocols")

The context owns the initialized session and closes it on exit, including on failure or cancellation. Native wrappers and wrap_tool() adapters share their parent's managed connection. Outside the context, they return to per-operation sessions. Discovery follows all tools/list pages and caches the complete list; use await adapter.list_tools_async(refresh=True) to refresh it.

For per-operation connections, use the agent helper directly:

await agent.add_mcp(transport="streamable_http", url="https://example.com/mcp")

Omitting transport with a URL retains legacy SSE behavior. The blocking agent.sync.add_mcp(...) accepts the same options.

Results, schemas and errors​

The adapter follows the MCP tool result contract.

ResponsePython return value
One plain text block without metadata or structured contentstr, including an empty string
Empty content without other dataNone
Structured data, multiple blocks, non-text content, or metadataSerializable dictionary with the complete MCP fields
isError=TrueRaises MCPToolError

Rich results preserve content, structuredContent, _meta, annotations, extension fields and null values. Access structured data through result["structuredContent"]. An empty structured object remains a structured result. Images, audio, resource links, and embedded resources retain their original content blocks.

Discovered tools retain their original input and output schemas, including nullable unions and local references. MCP arguments use JSON Schema validation without native Python coercion or changes to additionalProperties. output_schema describes the MCP structuredContent field, not the surrounding result dictionary. The MCP SDK validates structured outputs against the server's schema.

from protolink.tools.adapters import MCPToolError

try:
result = await agent.call_tool("remote_search", query="agent protocols")
except MCPToolError as error:
print(error.tool_name, error.result) # Complete failed MCP result

Direct calls raise MCPToolError; the existing task runtime records a failed tool output and convenience inference follows its existing failure behavior. No failed call is silently converted into successful text or automatically resubmitted.

Local MCP Server (stdio)​

Connect to a local MCP server running as a Python script:

from protolink.tools.adapters import MCPToolAdapter

# Connect to a local MCP server
adapter = MCPToolAdapter(transport="stdio", command="python", args=["path/to/mcp_server.py"])

# Or with a Node.js server
adapter = MCPToolAdapter(transport="stdio", command="node", args=["path/to/mcp_server.js"])

Remote MCP Server (SSE)​

Connect to a remote MCP server over HTTP:

from protolink.tools.adapters import MCPToolAdapter

# Connect to a remote MCP server
adapter = MCPToolAdapter(transport="sse", url="http://localhost:8080/sse")

# With authentication
adapter = MCPToolAdapter(
transport="sse", url="https://api.example.com/mcp/sse", headers={"Authorization": "Bearer your-api-token"}
)

Discovering Tools​

list_tools()​

Retrieve all available tools from the MCP server as dictionaries:

tools = adapter.list_tools()

for tool in tools:
print(f"Tool: {tool['name']}")
print(f" Description: {tool['description']}")
print(f" Input Schema: {tool['input_schema']}")
print(f" Input Types: {tool['input_types']}")
print(f" Callable: {tool['callable']}")

The returned dictionaries contain the MCP name, description, original input and output schemas, a shallow Python input-type mapping and a synchronous callable. Nullable, union, referenced and unsupported types use Any in the inspection mapping; validation still uses the original JSON Schema. See MCPToolAdapter.list_tools for caching and event-loop behavior.

get_tools()​

Retrieve all tools as BaseTool-compatible objects:

base_tools = adapter.get_tools()

for tool in base_tools:
print(f"{tool.name}: {tool.description}")
print(f" Input Schema: {tool.input_schema}")
# e.g., {"type": "object", "properties": {"location": {"type": "string"}}}

Returns a list of native Protolink Tool instances. Each tool:

  • Has name, description, input_schema populated from the MCP server
  • Has tags=["mcp"] to identify it as an MCP-sourced tool
  • Can be directly registered on a Protolink agent via agent.add_tool()

Display all available tools in a human-readable format:

adapter.print_tools()

Output:

🛠 Available MCP Tools:

🔹 Name : add
Description: Add two integers.
Input Schema: {'properties': {'a': {'type': 'integer'}, 'b': {'type': 'integer'}}, ...}
Input Types : {'a': <class 'int'>, 'b': <class 'int'>}

🔹 Name : greet
Description: Greet a person by name.
Input Schema: {'properties': {'name': {'type': 'string'}}, ...}
Input Types : {'name': <class 'str'>}

Invoking Tools​

There are multiple ways to invoke MCP tools, depending on whether you need synchronous or asynchronous execution:

Method 1: get_callable() - Synchronous Callable​

Get a synchronous callable for a specific tool. Best for quick scripts and non-async contexts:

# Get the synchronous callable
add = adapter.get_callable("add")

# Invoke with keyword arguments (no await needed)
result = add(a=5, b=7)
print(result) # "12"
Sync vs Async

get_callable() returns a synchronous function that uses asyncio.run() internally. This is simple but cannot be used inside an existing async context (it would cause a nested event loop error).

Get all tools as native Protolink Tool objects with async __call__ methods:

import asyncio

# Get all tools as native Tool objects
tools = adapter.get_tools()

# Find a specific tool
multiply = next(t for t in tools if t.name == "multiply")

# Invoke asynchronously
result = asyncio.run(multiply(a=5, b=7))
print(result) # "35"

This is the recommended approach for:

  • Registering tools on Protolink agents
  • Using tools in async contexts
  • Avoiding nested event loop issues
Agent Integration

get_tools() returns Tool objects that can be directly registered via agent.add_tool(tool). The agent's async runtime will properly await tool calls.

Method 3: wrap_tool() - Single BaseTool Instance​

Wrap a specific tool as a BaseTool-compatible object:

# Wrap the tool
add_tool = adapter.wrap_tool("add")

# Access metadata
print(add_tool.name) # "add"
print(add_tool.description) # "Add two integers."
print(add_tool.input_schema) # {"type": "object", "properties": {"a": {"type": "integer"}}, ...}

# Invoke asynchronously
import asyncio

result = asyncio.run(add_tool(a=5, b=7))
print(result) # "12"

Method 4: Via list_tools() Callable​

Use the synchronous callable directly from the tool dictionary:

tools = adapter.list_tools()

# Find the tool you want
add_tool = next(t for t in tools if t["name"] == "add")

# Invoke it (synchronous)
result = add_tool["callable"](a=5, b=7)
print(result) # "12"
Sync Callable Limitation

The callable in list_tools() is synchronous and cannot be used inside an async context. For async usage, use get_tools() instead.


Registering MCP Tools on Agents​

Once you have MCP tools, you can register them on a Protolink agent:

from protolink.agents import Agent
from protolink.models import AgentCard
from protolink.tools.adapters import MCPToolAdapter

# Create the agent
agent_card = AgentCard(url="http://localhost:8020", name="mcp_agent", description="Agent with MCP tools")
agent = Agent(card=agent_card, transport="http")

# Connect to MCP server
adapter = MCPToolAdapter(transport="stdio", command="python", args=["mcp_server.py"])

# Get all tools as native Protolink Tool objects
mcp_tools = adapter.get_tools()

# Register each tool with the agent
for tool in mcp_tools:
agent.add_tool(tool)
Native Tool Integration

get_tools() returns native Protolink Tool objects with tags=["mcp"], making them fully compatible with the agent system. No additional wrapping is needed.


Complete Example​

Here's a complete example showing how to create an MCP server and use it with Protolink:

The protolink[mcp] extra uses MCP SDK 2.2 or newer. Server examples use MCPServer, the SDK 2 replacement for FastMCP. Python SDK fields use snake_case; ProtoLink's rich result dictionaries retain the MCP wire names, including structuredContent and isError. In server tools, raise mcp.server.mcpserver.exceptions.ToolError for an expected error whose message should reach the caller; the SDK hides the details of unexpected exceptions.

MCP Server (mcp_server.py)​

from mcp.server.mcpserver import MCPServer

# Create the MCP server
mcp = MCPServer(name="math-tools", instructions="Simple MCP server with math tools")


@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two integers."""
return a + b


@mcp.tool()
def multiply(a: int, b: int) -> int:
"""Multiply two integers."""
return a * b


@mcp.tool()
def greet(name: str) -> str:
"""Greet a person by name."""
return f"Hello, {name}! 👋"


if __name__ == "__main__":
mcp.run()
from protolink.tools.adapters import MCPToolAdapter

# Connect to the MCP server
adapter = MCPToolAdapter(transport="stdio", command="python", args=["mcp_server.py"])

# Discover available tools
print("Available tools:")
adapter.print_tools()

# Get tools as BaseTool objects
tools = adapter.get_tools()
print(f"\nFound {len(tools)} tools")

# Use the add tool
add = adapter.get_callable("add")
result = add(a=10, b=20)
print(f"\n10 + 20 = {result}")

# Use the greet tool
greet = adapter.get_callable("greet")
message = greet(name="World")
print(f"\n{message}")

Output:

Available tools:

🛠 Available MCP Tools:

🔹 Name : add
Description: Add two integers.
...

🔹 Name : multiply
Description: Multiply two integers.
...

🔹 Name : greet
Description: Greet a person by name.
...

Found 3 tools

10 + 20 = 30

Hello, World! 👋

MCPToolAdapter API Reference​

MCPToolAdapter.list_tools​

methodprotolink.tools.adapters.MCPToolAdapter.list_tools
source
list_tools(
  *,
  refresh: bool = False,
) -> list[dict]

Discover tools synchronously and return metadata dictionaries. The first call opens an MCP session and caches the resulting list; later calls return the cached list object unless refresh=True.

Parameters

refreshbooldefault: False

Bypass cached discovery and replace it with a fresh server response.

Returns

namestr

MCP tool identifier.

descriptionstr

Server-provided description, normalized to an empty string when absent.

input_schemadict[str, Any]

Original MCP inputSchema, or an empty dictionary.

input_typesdict[str, type]

Shallow mapping from top-level JSON Schema types to Python classes for display and introspection. Unsupported shapes become Any.

output_schemadict | None

Original MCP output schema describing structuredContent, when supplied. The legacy output key is an alias for this value.

callableCallable[..., Any]

Synchronous closure for this tool. It uses asyncio.run() and opens a new MCP session per invocation. Use async tool wrappers inside a managed session.

Raises

RuntimeError

Calling this synchronous method inside an active event loop fails because it uses asyncio.run().

MCP or transport error

Subprocess startup, SSE connection, initialization, protocol and discovery failures propagate.

Mutable cache

The returned list and its dictionaries are the cached objects, not defensive copies. Treat them as read-only or use refresh=True to replace mutated cache state.

MCPToolAdapter.get_tool​

methodprotolink.tools.adapters.MCPToolAdapter.get_tool
source
get_tool(
  tool_name: str,
) -> dict | None

Find one discovered metadata dictionary by exact name. This is a linear search over list_tools() and therefore uses its cache and synchronous event-loop constraints.

Parameters

tool_namestrrequired

Exact case-sensitive MCP tool name.

Returns

tooldict | None

Cached metadata dictionary when found; otherwise None.

MCPToolAdapter.get_tools​

methodprotolink.tools.adapters.MCPToolAdapter.get_tools
source
get_tools() -> list[Tool]

Convert every discovered MCP definition into a native asynchronous Tool. Each call constructs a new wrapper list, while discovery metadata can come from the adapter cache.

Returns

toolslist[Tool]

Native tools with the original MCP input and output schemas and tags=["mcp"]. Async callables reuse an active managed session or open a session for the operation.

Recommended for Agent

Register these wrappers with agent.add_tool(). Their asynchronous call path is compatible with Agent execution and does not nest asyncio.run().

MCPToolAdapter.get_callable​

methodprotolink.tools.adapters.MCPToolAdapter.get_callable
source
get_callable(
  tool_name: str,
) -> Callable[..., Any]

Create a synchronous closure that invokes the named MCP tool. The name is not checked against discovery at construction; the MCP server validates it when the closure runs.

Parameters

tool_namestrrequired

Tool identifier sent to session.call_tool().

Returns

callableCallable[..., Any]

Keyword-only synchronous wrapper returning plain text, an empty result, or a complete rich MCP result dictionary. Tool failures raise MCPToolError.

Event loops

The closure uses asyncio.run(). Do not call it from an asynchronous handler or notebook cell with an active event loop; use get_tools() there.

MCPToolAdapter.wrap_tool​

methodprotolink.tools.adapters.MCPToolAdapter.wrap_tool
source
wrap_tool(
  tool_name: str,
) -> MCPToolAdapter

Discover one tool and return a new adapter configured to act as that asynchronous BaseTool. Connection settings and the metadata cache are shared by reference with the parent at wrapping time.

Parameters

tool_namestrrequired

Exact tool name that must already be discoverable from the MCP server.

Returns

wrappedMCPToolAdapter

New adapter with the MCP name, description, input schema and output schema. It shares the parent adapter's session context. Tags remain None.

Raises

ValueError

No discovered tool has the requested name.

MCPToolAdapter.call​

async methodprotolink.tools.adapters.MCPToolAdapter.__call__
source
async __call__(
  **kwargs,
) -> Any

Invoke the MCP tool represented by a wrapped adapter. A plain connection adapter has an empty name and cannot be called directly.

Parameters

**kwargsAny

Arguments sent unchanged to the MCP server. This adapter path does not run ProtoLink's native Tool.validate_args().

Returns

resultAny

Plain text string, None for empty content, or a rich MCP result dictionary preserving all content and metadata. Tool failures raise MCPToolError.

Raises

ValueError

The adapter does not wrap a named tool.

MCP or transport error

Session and remote tool failures propagate.

MCPToolAdapter.print_tools​

methodprotolink.tools.adapters.MCPToolAdapter.print_tools
source
print_tools() -> None

Print cached or freshly discovered names, descriptions, input schemas and shallow Python input types to standard output.

Returns

NoneNone

Output is written for human inspection; no formatted string is returned.


Best Practices​

Registering built-ins​

Register selectively, configure capabilities explicitly and use the Agent execution path. See Built-in Tools for policy, direct-call and serialization behavior.

Tool Design​

  1. Clear descriptions: Write descriptions that help the LLM understand when to use each tool
  2. Typed parameters: Use type hints for all parameters
  3. Error handling: Raise ToolRejectedError for correctable requests only when they made no changes; retain ordinary exceptions for execution failures.
  4. Single responsibility: Each tool should do one thing well

MCP Integration​

  1. Connection reuse: Create one MCPToolAdapter and reuse it for multiple tool calls
  2. Caching: Use list_tools() without refresh=True to leverage caching
  3. Error handling: Wrap tool calls in try/except for network failures
  4. Transport choice: Use stdio for local servers, streamable_http for remote services and sse for legacy endpoints

Agent Registration​

  1. Selective registration: Only register tools the agent actually needs
  2. Descriptive names: Use clear, action-oriented names like search_documents not do_search
  3. Tag organization: Use consistent tagging for related tools

See Also​