Tools

from tenxgraph.core.graph import ToolNode
from tenxgraph.core.state import ToolResult
from tenxgraph.core.graph.tool_node import UnsupportedToolParameterError, HAS_MCP, HAS_FASTMCP
from tenxgraph.utils import tool, get_tool_metadata, has_tool_decorator

ToolNode(tools, client=None, pass_user_info_to_mcp=False)

from tenxgraph.core.graph import Agent, ToolNode


def lookup_order(order_id: str) -> dict:
    """Look up an order by ID."""
    return {"order_id": order_id, "status": "shipped"}


def refund_order(order_id: str, amount: float) -> dict:
    """Refund an order."""
    return {"refund_id": "R-1", "amount": amount}


# Register local functions; each is exposed under its __name__
tool_node = ToolNode([lookup_order, refund_order])

# Inspect the JSON schemas the model will see
schemas = tool_node.all_tools_sync()
print(len(schemas))  # 2

# Add another tool after construction
def cancel_order(order_id: str) -> dict:
    """Cancel an order."""
    return {"order_id": order_id, "status": "cancelled"}

tool_node.add_tool(cancel_order)

# Give the tools to an agent; also add tool_node to your StateGraph as a node
agent = Agent(model="gpt-4o", provider="openai", tool_node=tool_node)

ToolResult(message=None, state=None, is_error=False, *, content=None)

from tenxgraph.core.state import ToolResult, AgentState

class MyState(AgentState):
    jd_name: str = ""

def update_context(state: MyState, jd_name: str) -> ToolResult:
    return ToolResult(
        message=f"JD name updated to '{jd_name}'",
        state={"jd_name": jd_name},
    )

@tool(
    _func=None,
    *,
    name: str | None = None,
    description: str | None = None,
    tags: list[str] | set[str] | None = None,
    provider: str | None = None,
    capabilities: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    parameters: dict[str, Any] | None = None,
)

from tenxgraph.core.graph import ToolNode
from tenxgraph.utils import tool

@tool(name="lookup_order", tags=["search", "orders"])
def find_order(order_id: str) -> dict:
    """Look up an order by ID."""
    return {"order_id": order_id, "status": "shipped"}

@tool(tags=["write", "payments"])
async def refund_order(order_id: str, amount: float) -> dict:
    """Refund an order."""
    return {"refund_id": "R-1", "amount": amount}

tool_node = ToolNode([find_order, refund_order])

get_tool_metadata(func: Callable) -> dict[str, Any]

from tenxgraph.utils import tool, get_tool_metadata

@tool(name="my_tool", tags=["test"])
def example():
    """Example function."""
    pass

metadata = get_tool_metadata(example)
print(metadata["name"])  # "my_tool"
print(metadata["tags"])  # {"test"}

has_tool_decorator(func: Callable) -> bool

from tenxgraph.utils import tool, has_tool_decorator

@tool
def decorated():
    pass

def not_decorated():
    pass

print(has_tool_decorator(decorated))      # True
print(has_tool_decorator(not_decorated))  # False

from tenxgraph.core.graph.tool_node import UnsupportedToolParameterError

from typing import TypedDict

from tenxgraph.core.graph import ToolNode


class Options(TypedDict):
    verbose: bool


def bad_tool(options: Options) -> str:
    """A tool whose parameter (a TypedDict) is not supported."""
    return "ok"


try:
    # Schemas are built from annotations; unsupported ones raise this error
    schemas = ToolNode([bad_tool]).all_tools_sync()
except UnsupportedToolParameterError as e:
    print(f"Cannot register tool: {e}")

from tenxgraph.core.graph import ToolNode
from tenxgraph.core.graph.tool_node import HAS_MCP, HAS_FASTMCP


def my_function(text: str) -> str:
    """Echo text."""
    return text


if HAS_FASTMCP and HAS_MCP:
    from fastmcp import Client

    client = Client("http://localhost:8000/mcp")  # replace with your MCP server
    tools = ToolNode([my_function], client=client)
else:
    print('MCP support requires: pip install "10xgraph[mcp]"')
    tools = ToolNode([my_function])