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.

Overview​

Protolink provides a flexible tool system with three approaches:

  • Built-in Tools: Opt-in, dependency-free factories for common read-only and pure operations
  • 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.builtinsImplementations for the dependency-free built-in tools
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 toolsExplicitly register dependency-free web search, safe URL fetch, calculator, and current-datetime tools.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, approval, cancellation, telemetry, and tool-call budget checks. Register it and use agent.call_tool(), agent.call_tool_in_context(), or task inference when those controls matter.

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,
)

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.

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.

Raises

ValueError | TypeError

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

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.

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 includes four dependency-free tool factories for common agent tasks:

  • web_search() creates web_search, which requires network.read and returns normalized ranked source snippets.
  • fetch_url() creates fetch_url, which requires network.read and returns bounded readable text from one public URL.
  • calculator() creates calculator, a pure bounded arithmetic evaluator with no protected capability.
  • current_datetime() creates current_datetime, a timezone-aware clock tool with no protected capability.

Factories return fresh native Tool instances. Nothing is enabled automatically: register only the capabilities an agent needs.

from protolink import Agent, AgentCard, CapabilityPolicy
from protolink.tools import calculator, current_datetime, fetch_url, web_search

agent = Agent(
    card=AgentCard(
        name="researcher",
        description="Finds and summarizes public information",
        url="runtime://researcher",
    ),
    transport="runtime",
    policy=CapabilityPolicy(
        {"network.read": "allow"},
        default_effect="deny",
    ),
)

agent.add_tool(web_search())
agent.add_tool(fetch_url())
agent.add_tool(calculator())
agent.add_tool(current_datetime())

Registered tools participate in schema validation, runtime policy, and AgentSkill advertising. When the inference loop invokes one during a task, the task's cancellation, telemetry, and tool-call budget controls apply as well. network.read identifies the authority required by web_search and fetch_url; calculator and current_datetime declare no protected capability.

Default policy and direct calls

The default CapabilityPolicy is allow-by-default for backward compatibility. Declaring network.read makes authority visible and configurable, but does not deny it by itself. Pass a restrictive policy when network access should be denied or approval-gated.

Calling a Tool object directly, such as await web_search()(query="..."), invokes the tool without the Agent and therefore bypasses Agent policy and approval. Use agent.call_tool(...) for Agent validation and policy, or let the inference loop invoke a registered tool when the task's full runtime controls should apply.

Agent dict/YAML serialization preserves each built-in's stable identity and the declarative rules, default effect, and name of ProtoLink's first-party CapabilityPolicy. Custom policy implementations and approval callbacks are executable application objects and are not embedded; pass them explicitly when restoring, for example Agent.from_yaml("agent.yaml", policy=custom_policy, approval_handler=approve). An explicit policy override takes precedence over serialized first-party policy data.

web_search() has one normalized result contract across three explicit engines:

  • engine="brave" is the default. It uses the Brave Search API and reads BRAVE_SEARCH_API_KEY from the environment only when invoked. The key is not captured by the Tool, stored in Agent configuration, or required merely to import or register the factory.
  • engine="duckduckgo" needs no API key or additional dependency. It reads DuckDuckGo's published non-JavaScript HTML search as a best-effort interface.
  • engine="wikipedia" needs no API key or additional dependency. It uses English Wikipedia's documented REST page-search API, which is the reliable keyless choice for encyclopedia and factual discovery. It supports freshness="any" only.

Engine selection is per call and there is no silent fallback. A missing Brave key therefore remains a clear configuration error instead of unexpectedly sending the query to another provider.

export BRAVE_SEARCH_API_KEY="your-key"
result = await agent.call_tool(
    "web_search",
    query="Python 3.14 release notes",
    max_results=5,
)

keyless_result = await agent.call_tool(
    "web_search",
    query="What is the capital of Greece?",
    engine="wikipedia",
)

best_effort_result = await agent.call_tool(
    "web_search",
    query="Python structured concurrency",
    engine="duckduckgo",
    freshness="month",
)

For a complete Agent-path CLI, see examples/builtin_web_search.py. It registers the built-in with an explicit network.read policy, supports all three engines, and prints the normalized JSON result:

# Keyless search through Wikipedia's documented API (example default)
python examples/builtin_web_search.py "What is the capital of Greece?"

# Documented Brave API
export BRAVE_SEARCH_API_KEY="your-key"
python examples/builtin_web_search.py "Python structured concurrency" --engine brave

# Keyless, best-effort DuckDuckGo HTML search
python examples/builtin_web_search.py "Python structured concurrency" --engine duckduckgo

Running the example without a query only prints its CLI help, so it is safe to inspect without credentials or a network request.

factoryprotolink.tools.web_search
source
web_search() -> Tool

Create a fresh Tool named web_search. The factory does not make a request and does not read the Brave credential; provider selection and credential lookup happen only when the returned tool is invoked.

Returns

toolTool

A native tool tagged builtin, web, search, and read-only, with the network.read capability and a bounded provider-neutral output schema.

Generated tool call

querystrrequired

Search text after surrounding whitespace is removed. It must contain 1–400 characters and no more than 50 whitespace-separated words.

max_resultsintdefault: 5

Maximum normalized results returned to the model. Accepted range: 1–10.

freshness"any" | "day" | "week" | "month" | "year"default: "any"

Optional result-age filter. Wikipedia accepts only "any"; requesting another value with that engine raises ValueError.

engine"brave" | "duckduckgo" | "wikipedia"default: "brave"

Explicit provider. Brave requires BRAVE_SEARCH_API_KEY; DuckDuckGo and Wikipedia are keyless and never selected as a silent fallback.

Returns from invocation

resultdict[str, Any]

Contains the normalized query, selected provider, ranked result objects, more_results_available, and untrusted_content=True. Each result includes title, URL, snippet, and an explicit sponsored marker.

Raises

ValueError

Invalid query length, word count, result limit, freshness, provider selection, or missing Brave credential.

RuntimeError

Provider response, content, challenge, HTTP, decoding, or bounded-transfer failures.

The tool normalizes all three engines into provider-neutral JSON-compatible data and bounds the result count and text placed into model context. Every result includes sponsored; Brave and Wikipedia results use False, while recognized DuckDuckGo advertisements stay in provider order with sponsored=True. Every engine uses a fixed HTTPS endpoint with DNS validation, a 2,000,000-byte response limit, a 10-second transport deadline, and no redirects. Wikipedia excerpts are converted from bounded provider markup to plain text. DuckDuckGo organic redirect links are decoded locally and validated; sponsored click URLs remain intact. Results also include the selected provider, more_results_available, and the explicit marker untrusted_content=True.

DuckDuckGo's HTML page is a human-facing interface rather than a versioned developer API. It can change markup, rate-limit automated requests, or return a human-verification challenge. ProtoLink does not spoof a browser, suppress or discard recognized advertising, retry a challenge, or attempt to bypass one; it raises a clear error that points to Wikipedia as the keyless alternative. Applications distributing a DuckDuckGo-backed integration should review DuckDuckGo's URL-parameter and partnership guidance. Use Wikipedia for reliable keyless encyclopedia search or Brave when a documented, general-web provider contract is required. With every engine, search queries leave the process, and titles, URLs, snippets, and page content are untrusted external data. Do not treat search output as instructions, executable content, or proof that a claim is correct.

URL Fetch​

fetch_url() retrieves bounded textual content from a public HTTP or HTTPS URL. It rejects credentials in URLs, non-HTTP schemes, and private, loopback, link-local, reserved, or otherwise non-public targets. Redirect destinations are resolved and validated again before they are followed. Responses are subject to redirect, timeout, byte, character, and supported-text-content limits; the result reports when extracted text was truncated.

page = await agent.call_tool("fetch_url", url="https://example.com/")
factoryprotolink.tools.fetch_url
source
fetch_url() -> Tool

Create a fresh Tool named fetch_url. Construction is side-effect free; DNS resolution and network access begin only when the returned tool is invoked.

Returns

toolTool

A native read-only web tool with the network.read capability, public-destination validation, bounded redirects and bytes, and an explicit output schema.

Generated tool call

urlstrrequired

Public HTTP or HTTPS URL of at most 2,048 characters. Embedded credentials, nonstandard ports, unsafe address ranges, HTTPS downgrades, and non-public redirect targets are rejected.

max_charsintdefault: 12000

Maximum readable text characters returned after download and decoding. Accepted range: 1–50,000; transfer bytes are bounded separately.

Returns from invocation

resultdict[str, Any]

Final validated URL, HTTP status, normalized content type, extracted title, bounded text, truncation flag, and untrusted_content=True.

Raises

ValueError

Invalid URL shape, scheme, credentials, port, address, redirect destination, or character limit.

RuntimeError

HTTP, redirect, timeout, response-size, content-type, charset, or HTML-decoding failures.

After each destination is DNS-validated, the transfer is limited to 1,000,000 bytes, four validated redirects, and a 10-second transport deadline for each request or redirect before the max_chars return bound is applied. DNS lookup uses the host operating system's resolver and is not included in that transport deadline. These restrictions reduce accidental server-side request forgery and context exhaustion; they do not make remote content trustworthy. Treat returned text as untrusted input and keep application-specific authorization at the Agent policy boundary.

Calculator and Current Datetime​

calculator() evaluates a deliberately small arithmetic grammar rather than Python code. It never uses eval, rejects names and function calls, and enforces expression-complexity, exponent, magnitude, and finite-result limits.

current_datetime() returns structured current-time data for the requested timezone. UTC works without a host timezone database; other IANA zones use the system database, or the tzdata package when a host does not provide one. Invalid or unavailable timezone identifiers raise a clear tool error rather than silently falling back to local machine time.

calculation = await calculator()(expression="(18 + 6) / 3")
now = await current_datetime()(timezone="Europe/Zurich")

calculator​

factoryprotolink.tools.calculator
source
calculator() -> Tool

Create a fresh pure arithmetic tool. The returned callable parses a restricted Python expression AST; it never uses eval and cannot resolve names, attributes, calls, booleans, or complex values.

Generated tool call

expressionstrrequired

Arithmetic expression of 1–256 characters using numbers, parentheses, unary signs, and +, -, *, /, //, %, or **. Syntax-tree size, exponent size, numeric magnitude, and finite-result limits prevent resource-heavy evaluation.

Returns

resultdict[str, int | float | str]

The trimmed original expression and its finite numeric result.

Raises

ValueError

Empty or invalid arithmetic, unsupported syntax, division by zero, oversized powers or values, excessive complexity, and non-finite results.

current_datetime​

factoryprotolink.tools.current_datetime
source
current_datetime() -> Tool

Create a fresh timezone-aware clock tool. UTC requires no external service or timezone database; other IANA identifiers are resolved through the host database or the optional tzdata package.

Generated tool call

timezonestrdefault: "UTC"

IANA timezone name of at most 100 characters. The tool never silently substitutes host-local time for an unknown zone.

Returns

resultdict[str, Any]

Requested timezone, ISO-8601 timestamp, date, time, weekday, UTC offset, and Unix timestamp.

Raises

ValueError

Empty, oversized, unknown, or unavailable timezone identifiers.

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​

To register a native tool, decorate an async function with @agent.tool:

from protolink.agents import Agent
from protolink.models import AgentCard

agent_card = AgentCard(
    url="http://localhost:8020",
    name="calculator_agent", 
    description="Agent with math tools"
)
agent = Agent(card=agent_card, transport="http")

@agent.tool(name="add", description="Add two numbers together")
async def add_numbers(a: int, b: int) -> int:
    """Add two integers and return the result."""
    return a + b


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

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

Agent.tool​

methodprotolink.agents.Agent.tool
source
tool(
  name: str,
  description: str,
  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,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Create a decorator that wraps a Python callable in Tool, registers it immediately on this Agent, and synchronizes the corresponding advertised AgentSkill. The decorated name remains bound to the original function, while the runtime wrapper is available through agent.tools[name].

Parameters

namestrrequired

Stable identifier exposed to models, clients, task parts, and policy actions. Reusing an existing runtime name replaces the tool and its matching skill.

descriptionstrrequired

Selection guidance shown to the LLM and discovery clients. Include the operation's purpose, prerequisites, and important side effects.

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

decoratorCallable

A decorator that registers the wrapped function and then returns that original function unchanged.

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 return a structured tool error instead of reaching user code.

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) when a deterministic application path invokes a tool directly and needs the same per-run permissions, cancellation, and approval behavior as task execution.

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 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
sseServer-Sent Events over HTTPRemote 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 discovery or invocation operation creates and initializes a session for that operation.

Parameters

transportstrdefault: "stdio"

MCP client transport. Supported values are "stdio" for a local subprocess and "sse" for a remote Server-Sent Events endpoint.

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

SSE endpoint required when the first sse operation runs.

headersdict[str, str] | Nonedefault: None

Headers forwarded by the SSE 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 SSE URL. Configuration is validated lazily, not by the constructor.

Session lifetime

The current adapter does not keep one MCP session open across calls. It caches discovered metadata, but each uncached discovery or tool invocation opens, initializes, and closes its own stdio or SSE session.

Connecting to MCP Servers​

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, input schema, shallow Python input-type mapping, an output placeholder, and a synchronous callable. See MCPToolAdapter.list_tools for the exact result contract, caching behavior, and event-loop limitation.

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).

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
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:

MCP Server (mcp_server.py)​

from mcp.server.fastmcp import FastMCP

# Create the MCP server
mcp = FastMCP(
    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.

outputNone

Reserved placeholder; the current adapter does not expose MCP output schemas.

callableCallable[..., Any]

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

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 MCP name, description, input schema, output_schema=None, and tags=["mcp"]. Their async callables open a fresh MCP session for each invocation.

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 the first text content item when present, otherwise None.

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 name, description, and input_schema populated. output_schema and 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

Text from the first MCP content item when it has a text attribute; otherwise None.

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​

Built-in Tools​

  1. Register selectively: Built-ins are opt-in; add only the tools an agent needs.
  2. Configure policy: Use CapabilityPolicy to allow, deny, or approval-gate network.read explicitly.
  3. Treat external data as untrusted: Search results and fetched pages can contain incorrect or adversarial text.
  4. Use the Agent execution path: Direct Tool calls are convenient for low-level tests but bypass Agent runtime controls.

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 clear exceptions for invalid inputs
  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, sse for remote services

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​