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
| Outcome | Contract | Reporting |
|---|---|---|
| Successful execution | The tool returned its result | tool_result / action.completed |
| Request rejected | ToolRejectedError, or pre-execution tool/schema validation; nothing changed in this request | tool_rejected / action.rejected |
| Delegation rejected | Unsupported advertised target/tool/schema, or a correlated single-tool no-change response | agent_call_rejected / action.rejected |
| Permission denied | Policy or approval denied the action | action.denied |
| Execution failed | Ordinary exception inside the tool; effects may be partial or unknown | tool_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
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
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
| Module | Description |
|---|---|
protolink.tools | Core interfaces, native implementation and public built-in factories |
protolink.tools.builtins | Built-in factories and configurable backends |
protolink.tools.adapters | Adapters for integrating external tool systems |
The callable capability layer for native Python functions, MCP-backed tools, JSON schemas, examples, capability policies and approval-aware action metadata.
protolink.toolsagent.add_tool(web_search())BaseTool@agent.toolMCPToolAdaptercapabilitiesBaseTool 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
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) -> AnyThe 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
namestrStable 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.descriptionstrHuman-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] | NoneJSON Schema for accepted keyword arguments. Agent execution validates model-provided arguments against this schema before the callable runs.
output_schemaAny | NoneOptional schema describing the returned value. It is advertised to callers but does not currently validate the runtime result.
tagslist[str] | NoneDiscovery and presentation labels copied to the Agent's skill card.
exampleslist[Any] | NoneRepresentative calls or values copied to
AgentSkill.examples. They guide clients and models but are not executed automatically.capabilitiesCollection[str] | NoneAuthority strings merged into the
RunActionevaluated immediately before execution. Capabilities become enforceable only when the call passes through Agent policy.
Call
**kwargsAnyKeyword arguments matching
input_schema. Positional invocation is outside the protocol.returnAnyTool-specific result. Implementations may return any JSON-compatible or application value.
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
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
namestrrequiredStable runtime and advertised identifier.
descriptionstrrequiredPurpose presented to models and discovery clients.
input_schemadict[str, Any] | NonerequiredExplicit JSON Schema or legacy field map. Pass
Noneto infer an object schema fromfunc's parameters and type annotations.output_schemaAny | NonerequiredExplicit result schema, or
Noneto infer one from the callable's return annotation.tagslist[str] | NonerequiredDiscovery labels.
Nonebecomes an empty list.funcCallable[..., Any]requiredWrapped Python function. Synchronous return values are accepted; awaitable results are awaited automatically.
argsdict[str, Any] | Nonedefault: NoneLegacy metadata retained on the dataclass. Runtime invocation uses arguments passed to
call(), not this mapping.exampleslist[Any] | Nonedefault: NoneAdvertised examples.
Nonebecomes an empty list.capabilitiesCollection[str] | Nonedefault: NoneRequired policy capabilities. Empty values are removed and duplicates are discarded while preserving first occurrence.
action_builderActionBuilder | Nonedefault: NoneOptional sync or async callback that receives validated arguments and the active
RunContext, then returns a customizedRunActionwith preview artifacts or metadata.sync_executionLiteral["inline", "thread"] | Nonedefault: NoneOverride synchronous execution for this tool;
Noneinherits the executing agent's default. Standalone tools default to inline execution. Async callables are awaited directly in either mode.
Raises
ValueError | TypeErrorCallable inspection, type-hint resolution, or explicit schema normalization can fail during construction.
Tool.from_callable
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,
) -> ToolCreate 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
funcare 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
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] | NonerequiredUntrusted argument mapping.
Nonebecomes 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
ValueErrorSchema violations, missing or unexpected fields, incompatible values and annotation-validation failures.
Tool.call
async __call__(
**kwargs: Any,
) -> AnyValidate 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
**kwargsAnyValues accepted by the generated or explicit input schema and the callable signature.
Returns
resultAnyDirect or awaited result from
func.output_schemaadvertises this value but is not enforced here.
Raises
ValueErrorArgument validation fails before user code executes.
callable errorExceptions from the wrapped function propagate unchanged.
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 prepare_action(
arguments: dict[str, Any],
context: RunContext,
) -> RunActionBuild 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]requiredAlready validated keyword arguments proposed for execution.
contextRunContextrequiredActive run identity, permissions, session, cancellation and budget context supplied to a custom builder.
Returns
actionRunActionDomain-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
TypeErrorThe configured
action_builderdoes not returnRunAction.builder errorExceptions raised by the application builder propagate to the execution layer.
ToolRejectedErrorExplicit 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
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: NoneStable 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: NoneSelection 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: NoneExplicit JSON Schema or legacy field map.
Noneinfers a schema from the decorated function's signature and type hints.output_schemadict[str, Any] | Nonedefault: NoneExplicit result schema.
Noneinfers it from the return annotation; it is advertised but does not validate the returned runtime value.tagslist[str] | Nonedefault: NoneDiscovery labels copied to the generated
AgentSkill.exampleslist[Any] | Nonedefault: NoneRepresentative examples copied to the skill card.
capabilitieslist[str] | tuple[str, ...] | set[str] | Nonedefault: NoneAuthority required before Agent execution. The wrapper normalizes non-empty values and merges them into every prepared action.
action_builderActionBuilder | Nonedefault: NoneOptional 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 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
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
schemaAnyrequiredSupported schema representation.
Nonebecomes an empty closed object schema; dictionaries that already look like JSON Schema are copied before normalization.titlestr | Nonedefault: NoneOptional 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
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]requiredFunction whose signature and resolved type hints describe tool input.
titlestrrequiredRequired schema title, normally derived from the tool name.
Returns
schemadict[str, Any]JSON Schema object with
additionalProperties=False.
infer_output_schema
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]requiredCallable whose return type should be advertised.
titlestrrequiredTitle added to the returned schema.
Returns
schemadict[str, Any]JSON Schema describing the annotated return value.
validate_tool_args
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] | NonerequiredProposed arguments.
Noneis normalized to an empty dictionary and the caller's mapping is copied.input_schemadict[str, Any] | NonerequiredSchema used for structural validation and conservative coercion of strings, numbers, booleans, arrays and nested objects.
type_hintsdict[str, Any] | Nonedefault: NoneResolved annotations keyed by parameter name. When supplied, matching values are validated and reconstructed after schema checks.
signatureinspect.Signature | Nonedefault: NoneCallable signature used to detect required and unexpected keyword fields. A callable accepting
**kwargspermits additional names.
Returns
argumentsdict[str, Any]New validated mapping, potentially containing coerced scalars or reconstructed annotated objects.
Raises
ValueErrorSchema 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
| Transport | Description | Use Case |
|---|---|---|
stdio | Local subprocess via stdin/stdout | Local Python/Node.js MCP servers |
streamable_http | Streamable HTTP | Remote MCP web services |
sse | Legacy Server-Sent Events over HTTP | Older MCP web services |
Constructor
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: NoneExecutable launched for
stdio, such as"python","node", or an MCP server binary. It is required when the first stdio operation runs.argslist[str] | Nonedefault: NoneArguments passed unchanged to the stdio command.
Nonebecomes an empty list.urlstr | Nonedefault: NoneEndpoint URL required for either HTTP transport.
headersdict[str, str] | Nonedefault: NoneHeaders forwarded by the selected HTTP client, commonly for authentication.
Nonebecomes an empty dictionary.
Raises
ImportErrorImporting
protolink.tools.adaptersfails when the optional MCP dependency is unavailable. Installprotolink[mcp].ValueErrorDiscovery or invocation raises for an unknown transport, missing stdio command, or missing HTTP URL. Configuration is validated lazily, not by the constructor.
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.
| Response | Python return value |
|---|---|
| One plain text block without metadata or structured content | str, including an empty string |
| Empty content without other data | None |
| Structured data, multiple blocks, non-text content, or metadata | Serializable dictionary with the complete MCP fields |
isError=True | Raises 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_schemapopulated 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()
print_tools()
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"
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).
Method 2: get_tools() - Native Protolink Tools (Async)
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
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"
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)
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()
Protolink Client
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
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: FalseBypass cached discovery and replace it with a fresh server response.
Returns
namestrMCP tool identifier.
descriptionstrServer-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 | NoneOriginal MCP output schema describing structuredContent, when supplied. The legacy
outputkey 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
RuntimeErrorCalling this synchronous method inside an active event loop fails because it uses
asyncio.run().MCP or transport errorSubprocess startup, SSE connection, initialization, protocol and discovery failures propagate.
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
get_tool(
tool_name: str,
) -> dict | NoneFind 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_namestrrequiredExact case-sensitive MCP tool name.
Returns
tooldict | NoneCached metadata dictionary when found; otherwise
None.
MCPToolAdapter.get_tools
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.
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
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_namestrrequiredTool 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.
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
wrap_tool(
tool_name: str,
) -> MCPToolAdapterDiscover 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_namestrrequiredExact tool name that must already be discoverable from the MCP server.
Returns
wrappedMCPToolAdapterNew adapter with the MCP name, description, input schema and output schema. It shares the parent adapter's session context. Tags remain
None.
Raises
ValueErrorNo discovered tool has the requested name.
MCPToolAdapter.call
async __call__(
**kwargs,
) -> AnyInvoke the MCP tool represented by a wrapped adapter. A plain connection adapter has an empty name and cannot be called directly.
Parameters
**kwargsAnyArguments sent unchanged to the MCP server. This adapter path does not run ProtoLink's native
Tool.validate_args().
Returns
resultAnyPlain text string,
Nonefor empty content, or a rich MCP result dictionary preserving all content and metadata. Tool failures raiseMCPToolError.
Raises
ValueErrorThe adapter does not wrap a named tool.
MCP or transport errorSession and remote tool failures propagate.
MCPToolAdapter.print_tools
print_tools() -> NonePrint cached or freshly discovered names, descriptions, input schemas and shallow Python input types to standard output.
Returns
NoneNoneOutput 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
- Clear descriptions: Write descriptions that help the LLM understand when to use each tool
- Typed parameters: Use type hints for all parameters
- Error handling: Raise
ToolRejectedErrorfor correctable requests only when they made no changes; retain ordinary exceptions for execution failures. - Single responsibility: Each tool should do one thing well
MCP Integration
- Connection reuse: Create one
MCPToolAdapterand reuse it for multiple tool calls - Caching: Use
list_tools()withoutrefresh=Trueto leverage caching - Error handling: Wrap tool calls in try/except for network failures
- Transport choice: Use
stdiofor local servers,streamable_httpfor remote services andssefor legacy endpoints
Agent Registration
- Selective registration: Only register tools the agent actually needs
- Descriptive names: Use clear, action-oriented names like
search_documentsnotdo_search - Tag organization: Use consistent tagging for related tools
See Also
- Agent Documentation - How agents use tools
- LLM Documentation - How LLMs invoke tools via
infer() - MCP Specification - Model Context Protocol details