universal-tool-server

from typing import Annotated
from starlette.requests import Request

from universal_tool_server.tools import InjectedRequest
from universal_tool_server import Server, Auth

app = Server()
auth = Auth()
app.add_auth(auth)


@auth.authenticate
async def authenticate(authorization: str) -> dict:
    """Authenticate incoming requests."""
    api_key = authorization

    # Replace this with actual authentication logic.
    api_key_to_user = {
        "1": {"permissions": ["authenticated", "group1"], "identity": "some-user"},
        "2": {"permissions": ["authenticated", "group2"], "identity": "another-user"},
    }
    # This is just an example. You should replace this with an actual
    # implementation.
    if not api_key or api_key not in api_key_to_user:
        raise auth.exceptions.HTTPException(detail="Not authorized")
    return api_key_to_user[api_key]


# Define tools

@app.add_tool(permissions=["group1"])
async def echo(msg: str) -> str:
    """Echo a message."""
    return msg + "!"


# Tool that has access to the request object
@app.add_tool(permissions=["authenticated"])
async def who_am_i(request: Annotated[Request, InjectedRequest]) -> str:
    """Get the user identity."""
    return request.user.identity


# You can also expose existing LangChain tools!
from langchain_core.tools import tool


@tool()
async def say_hello() -> str:
    """Say hello."""
    return "Hello"


# Add an existing LangChain tool to the server with permissions!
app.add_tool(say_hello, permissions=["group2"])

import asyncio

from universal_tool_client import get_async_client


async def main():
    if len(sys.argv) < 2:
        print(
            "Usage: uv run client.py url of universal-tool-server  (i.e. http://localhost:8080/)>"
        )
        sys.exit(1)

    url = sys.argv[1]
    client = get_async_client(url=url)
    # Check server status
    print(await client.ok())  # "OK"
    print(await client.info())  # Server version and other information

    # List tools
    print(await client.tools.list())  # List of tools
    # Call a tool
    print(await client.tools.call("add", {"x": 1, "y": 2}))  # 3

    # Get as langchain tools
    select_tools = ["echo", "add"]
    tools = await client.tools.as_langchain_tools(select_tools)
    # Async
    print(await tools[0].ainvoke({"msg": "Hello"}))  # "Hello!"
    print(await tools[1].ainvoke({"x": 1, "y": 3}))  # 4


if __name__ == "__main__":
    import sys

    asyncio.run(main())

from universal_tool_client import get_sync_client

from open_tool_server import Server
from langchain_core.tools import tool

app = Server()

# Say you have some existing langchain tool
@tool()
async def say_hello() -> str:
    """Say hello."""
    return "Hello"

# This is how you expose it via the API
app.tool(
    say_hello,
    # You can include permissions if you're setting up Auth
    permissions=["group2"]
)

import os

from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent

from universal_tool_client import get_sync_client

if "ANTHROPIC_API_KEY" not in os.environ:
    raise ValueError("Please set ANTHROPIC_API_KEY in the environment.")

tool_server = get_sync_client(
    url=... # URL of the tool server
    # headers=... # If you enabled auth
)
# Get tool definitions from the server
tools = tool_server.tools.as_langchain_tools()
print("Loaded tools:", tools)

model = ChatAnthropic(model="claude-3-5-sonnet-20240620")
agent = create_react_agent(model, tools=tools)
print()

user_message = "What is the temperature in Paris?"
messages = agent.invoke({"messages": [{"role": "user", "content": user_message}]})[
    "messages"
]

for message in messages:
    message.pretty_print()

from universal_tool_server import Server

app = Server(enable_mcp=True)


@app.add_tool()
async def echo(msg: str) -> str:
    """Echo a message."""
    return msg + "!"

from mcp import ClientSession

from mcp.client.sse import sse_client

async def main() -> None:
    # Please replace [host] with the actual host
    # IMPORTANT: Add /mcp/sse to the url!
    url = "[host]/mcp/sse" 
    async with sse_client(url=url) as streams:
        async with ClientSession(streams[0], streams[1]) as session:
            await session.initialize()
            tools = await session.list_tools()
            print(tools)
            result = await session.call_tool("echo", {"msg": "Hello, world!"})
            print(result)

@app.add_tool()
async def add(x: int, y: int) -> int:
    """Add two numbers."""
    return x + y

@app.add_tool(permissions=["group1"])
async def add(x: int, y: int) -> int:
    """Add two numbers."""
    return x + y

from typing import Annotated
from universal_tool_server import InjectedRequest
from starlette.requests import Request


@app.add_tool(permissions=["group1"])
async def who_am_i(request: Annotated[Request, InjectedRequest]) -> str:
    """Return the user's identity"""
    # The `user` attribute can be used to retrieve the user object.
    # This object corresponds to the return value of the authentication function.
    return request.user.identity

from universal_tool_client import get_async_client

async def get_tools():
    # Headers are entirely dependent on how you implement your authentication
    # (see Auth section)
    client = get_async_client(url="http://localhost:8080/", headers={"authorization": "api key"})
    tools = await client.tools.list()
    # If you need langchain tools you can use the as_langchain_tools method
    langchain_tools = await client.tools.as_langchain_tools()
    # Do something
    ...

from universal_tool_server import Auth

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict[bytes, bytes]) -> dict:
    """Authenticate incoming requests."""
    is_authenticated = ... # Your authentication logic here
    if not is_authenticated:
        raise auth.exceptions.HTTPException(detail="Not authorized")
    
    return {
        "identity": "some-user",
        "permissions": ["authenticated", "group1"],
        # Add any other user information here
        "foo": "bar",
    }