टूल्स

from dedalus_mcp import MCPServer, tool

@tool(description="दो संख्याएँ जोड़ें")
def add(a: int, b: int) -> int:
    return a + b

server = MCPServer("math")
server.collect(add)

import anyio
from dedalus_mcp import tool

@tool(description="उपयोगकर्ता डेटा प्राप्त करें (सिम्युलेटेड I/O)")
async def get_user(user_id: str) -> dict:
    await anyio.sleep(0.1)
    return {"user_id": user_id, "status": "ok"}

from typing import Literal
from pydantic import BaseModel
from dedalus_mcp import tool

class SearchFilters(BaseModel):
    category: str | None = None
    min_price: float = 0.0

@tool(description="उत्पादों को खोजें")
def search(
    query: str,
    limit: int = 10,
    sort: Literal["relevance", "price", "date"] = "relevance",
    filters: SearchFilters | None = None,
) -> list[dict]:
    return [{"query": query, "limit": limit, "sort": sort, "filters": filters.model_dump() if filters else None}]

from dedalus_mcp import tool

@tool(
    name="find_products",           # tool का नाम ओवरराइड करें
    description="Search catalog",   # tool का विवरण
    tags={"search", "catalog"},     # फ़िल्टरिंग/मेटाडेटा के लिए
)
def search_products_impl(query: str) -> list[dict]:
    return [{"id": "p_1", "name": "Widget", "query": query}]

from dedalus_mcp import tool

@tool(description="टेक्स्ट का विश्लेषण करें")
def analyze(text: str) -> dict:
    return {"word_count": len(text.split()), "char_count": len(text)}

from dedalus_mcp import tool
from dedalus_mcp.types import CallToolResult, TextContent

@tool(description="Custom result")
def custom() -> CallToolResult:
    return CallToolResult(
        content=[TextContent(type="text", text="Custom message")],
        isError=False,
    )

import anyio
from dedalus_mcp import tool, get_context

@tool(description="प्रगति रिपोर्टिंग के साथ फ़ाइलों को प्रोसेस करें")
async def process_files(paths: list[str]) -> dict:
    ctx = get_context()
    await ctx.info("प्रारंभ हो रहा है", data={"count": len(paths)})

    processed = 0
    try:
        async with ctx.progress(total=len(paths)) as tracker:
            for path in paths:
                # कार्य का अनुकरण; रद्दीकरण को टास्क रद्दीकरण के रूप में प्रदान किया जाता है
                await anyio.sleep(0.01)
                processed += 1
                await tracker.advance(1)
    except anyio.get_cancelled_exc_class():
        await ctx.warning("रद्द कर दिया गया", data={"processed": processed})
        raise

    return {"processed": processed}

from dedalus_mcp import MCPServer, tool

@tool(description="जोड़ना")
def add(a: int, b: int) -> int:
    return a + b

@tool(description="गुणा करना")
def multiply(a: int, b: int) -> int:
    return a * b

server = MCPServer("gated")
server.collect(add, multiply)
server.allow_tools({"add"})

from dedalus_mcp import tool

@tool(description="Divide")
def divide(a: float, b: float) -> float:
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

def test_add():
    assert add(2, 3) == 5