Agents & tools¶

import sys

import dotenv
from loguru import logger

dotenv.load_dotenv("../.env")

logger.remove()
_ = logger.add(sys.stderr, level="WARNING")

from pydantic import BaseModel

from kavalai import FunctionKernel, pythontool


class Resident(BaseModel):
    name: str
    born: str
    street: str


VILLAGE = {
    "Agnes Whitlow": Resident(
        name="Agnes Whitlow", born="1929-06-02", street="Willow Lane"
    ),
    "Thomas Cook": Resident(
        name="Thomas Cook", born="1994-04-12", street="Main Road"
    ),
    "Greta Lindqvist": Resident(
        name="Greta Lindqvist", born="1968-11-27", street="Cobbler's Path"
    ),
}


@pythontool
def find_resident(name: str) -> Resident:
    """Look up a Green Village resident by name."""
    return VILLAGE[name]


@pythontool
def village_population() -> int:
    """Return the number of residents of Green Village."""
    return 104


# A @pythontool function is unchanged — you can still call it directly.
print("direct call:", find_resident("Agnes Whitlow"))

kernel = FunctionKernel()
kernel.register_python_tool("find_resident", find_resident)
kernel.register_python_tool("village_population", village_population)

# Through the kernel, each tool is addressed by URI.
resident = await kernel.call_tool(
    "python://find_resident", {"name": "Greta Lindqvist"}
)
population = await kernel.call_tool("python://village_population")

print("via kernel :", resident)
print("street     :", resident.street)
print("population :", population.result)

@pythontool
def years_since(year: int) -> int:
    """Return how many years have passed since the given year (as of 2026)."""
    return 2026 - year


kernel.register_python_tool("years_since", years_since)

input_model = kernel.get_input_model("python://years_since")
output_model = kernel.get_output_model("python://years_since")
print("input fields :", list(input_model.model_fields))
print("output fields:", list(output_model.model_fields))

# The string "1929" is coerced to an int before the function runs.
age = await kernel.call_tool("python://years_since", {"year": "1929"})
print('years_since("1929") =', age.result)

print(await kernel.get_tool_descriptions())

from kavalai import Agent, make_client


class ResidentReport(BaseModel):
    name: str
    street: str
    age: int


agent = Agent(llm_client=make_client("openai/gpt-5.6-luna"), kernel=kernel)

report = await agent.prompt(
    "Look up Agnes Whitlow of Green Village. Then, using her birth "
    "year from that lookup, work out how old she is. Report her "
    "name, street and age.",
    response_model=ResidentReport,
    max_steps=5,
)
print(report)

import json

restricted = await kernel.get_tool_descriptions(
    allowed_tools=["python://village_population"]
)
print("visible to a restricted agent:",
      [t["name"] for t in json.loads(restricted)])

readonly_agent = Agent(
    llm_client=make_client("openai/gpt-5.6-luna"),
    kernel=kernel,
    allowed_tools=["python://village_population"],
)
print(await readonly_agent.prompt(
    "How many people live in Green Village?", max_steps=3
))

from kavalai import RestServer

kernel.register_rest_server(
    RestServer(name="weather", url="https://api.open-meteo.com/v1")
)
kernel.register_rest_tool(
    server_name="weather",
    tool_name="forecast",
    method="GET",
    input_schema={
        "type": "object",
        "properties": {
            "latitude": {"type": "number"},
            "longitude": {"type": "number"},
            "current": {"type": "string", "description": "e.g. temperature_2m"},
        },
        "required": ["latitude", "longitude"],
    },
    output_schema={
        "type": "object",
        "properties": {"current": {"type": "object"}},
    },
    description="Current weather for a GPS location from Open-Meteo.",
)

# The coordinates below are Tallinn's.
weather = await kernel.call_tool(
    "rest://weather.forecast",
    {"latitude": 59.437, "longitude": 24.7536, "current": "temperature_2m"},
)
print(weather.current)

weather_agent = Agent(
    llm_client=make_client("openai/gpt-5.6-luna"),
    kernel=kernel,
    allowed_tools=["rest://weather.*"],
)
print(await weather_agent.prompt(
    "What is the current temperature in Tallinn "
    "(59.437 N, 24.7536 E)?",
    max_steps=3,
))

import sys
import tempfile
import textwrap
from pathlib import Path

server_path = Path(tempfile.gettempdir()) / "green_village_mcp.py"
server_path.write_text(textwrap.dedent('''
    from mcp.server.fastmcp import FastMCP

    mcp = FastMCP("green-village")

    @mcp.tool()
    def honey_per_hive(total_tons: float, hives: int) -> float:
        """Kilograms of honey produced per beehive per year."""
        return (total_tons * 1000) / hives

    @mcp.tool()
    def loaves_per_resident(loaves_per_week: int, residents: int) -> float:
        """Weekly loaves baked per resident."""
        return loaves_per_week / residents

    if __name__ == "__main__":
        mcp.run()
'''))

from kavalai import McpServer

mcp_kernel = FunctionKernel()
mcp_kernel.register_mcp_server(
    McpServer(name="village", command=sys.executable, args=[str(server_path)])
)

# Registration only records the server. Connecting starts the process and asks
# it what it offers — so a misconfigured server fails here, before any tokens
# are spent on a run that was never going to work.
await mcp_kernel.connect_mcp_servers()

honey = await mcp_kernel.call_tool(
    "mcp://village.honey_per_hive", {"total_tons": 8, "hives": 26}
)
print("kg of honey per hive:", round(honey.result, 1))

# Connected, so the server's tools are described to the model.
print(await mcp_kernel.get_tool_descriptions())

mcp_agent = Agent(
    llm_client=make_client("openai/gpt-5.6-luna"), kernel=mcp_kernel
)

print(await mcp_agent.prompt(
    "Use the available tools to work out, for Green Village: "
    "kilograms of honey per beehive (8 tons a year from 26 hives) "
    "and loaves per resident (340 loaves a week for 104 residents). "
    "Round each to one decimal.",
    max_steps=4,
))

# MCP servers are subprocesses — shut them down when you are done.
# A notebook runs every cell as its own task, and the MCP transport must be
# closed by the task that opened it, so the kernel logs a warning here; the
# subprocess is stopped all the same. An application connects and closes in
# one place (`engine.connect()` / `engine.aclose()`), where it is silent.
await mcp_kernel.close()

from kavalai.tools.webtools.http_client import http_request

web_kernel = FunctionKernel()
web_kernel.register_python_tool("http.request", http_request)

response = await web_kernel.call_tool(
    "python://http.request",
    {
        "method": "GET",
        "url": "https://api.open-meteo.com/v1/forecast",
        "params": {
            "latitude": 59.437,
            "longitude": 24.7536,
            "current": "temperature_2m",
        },
    },
)
print("HTTP", response.status_code, "->", response.json_data["current"])