MCP con Python: permite que una IA consulte tu inventario

mkdir stockbridge-mcp
cd stockbridge-mcp

py -m venv .venv

mcp==2.3.0

.\.venv\Scripts\python.exe -m pip install -r requirements.txt

.\.venv\Scripts\python.exe -m pip check

[
  {
    "id": "P001",
    "nombre": "Café molido 500 g",
    "precio_cop": 24000,
    "stock": 3
  },
  {
    "id": "P002",
    "nombre": "Miel 250 g",
    "precio_cop": 18000,
    "stock": 0
  },
  {
    "id": "P003",
    "nombre": "Chocolate 250 g",
    "precio_cop": 12000,
    "stock": 5
  },
  {
    "id": "P004",
    "nombre": "Panela 500 g",
    "precio_cop": 6000,
    "stock": 12
  },
  {
    "id": "P005",
    "nombre": "Café en grano 500 g",
    "precio_cop": 28000,
    "stock": 2
  }
]

"""StockBridge: inventario de ejemplo consultable mediante MCP local."""

import json
from pathlib import Path
from typing import TypedDict

from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError


class Producto(TypedDict):
    id: str
    nombre: str
    precio_cop: int
    stock: int


class Busqueda(TypedDict):
    productos: list[Producto]
    total: int


class BajoStock(Busqueda):
    limite: int


def cargar_inventario(ruta: Path | None = None) -> list[Producto]:
    """Carga y valida el JSON antes de atender cualquier consulta."""
    ruta = ruta or Path(__file__).resolve().with_name("inventario.json")
    try:
        datos = json.loads(ruta.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"No se pudo cargar {ruta}: {exc}") from exc
    if not isinstance(datos, list):
        raise ValueError("inventario.json debe contener una lista de productos")

    ids: set[str] = set()
    for numero, producto in enumerate(datos, start=1):
        if not isinstance(producto, dict) or set(producto) != {"id", "nombre", "precio_cop", "stock"}:
            raise ValueError(f"Producto {numero}: se esperan id, nombre, precio_cop y stock")
        if (not isinstance(producto["id"], str) or not producto["id"].strip()
                or producto["id"] in ids
                or not isinstance(producto["nombre"], str) or not producto["nombre"].strip()
                or type(producto["precio_cop"]) is not int or producto["precio_cop"] < 0
                or type(producto["stock"]) is not int or producto["stock"] < 0):
            raise ValueError(f"Producto {numero}: id/nombre inválido, id duplicado o precio/stock inválido")
        ids.add(producto["id"])
    return datos


INVENTARIO = cargar_inventario()
mcp = MCPServer("StockBridge", version="0.1.0")


@mcp.tool()
def buscar_producto(nombre: str) -> Busqueda:
    """Busca productos por parte del nombre, sin distinguir mayúsculas; no modifica datos."""
    if not isinstance(nombre, str) or not nombre.strip():
        raise ToolError("nombre debe ser un texto no vacío")
    encontrados = [p for p in INVENTARIO if nombre.strip().casefold() in p["nombre"].casefold()]
    return {"productos": encontrados, "total": len(encontrados)}


@mcp.tool()
def productos_por_agotarse(limite: int = 5) -> BajoStock:
    """Lista productos con stock estrictamente menor al límite, incluyendo agotados."""
    if type(limite) is not int or limite < 0:
        raise ToolError("limite debe ser un entero mayor o igual que cero")
    encontrados = sorted((p for p in INVENTARIO if p["stock"] < limite),
                         key=lambda p: (p["stock"], p["id"]))
    return {"productos": encontrados, "total": len(encontrados), "limite": limite}


if __name__ == "__main__":
    mcp.run(transport="stdio")

.\.venv\Scripts\python.exe -c "from server import buscar_producto, productos_por_agotarse; print(buscar_producto('CAFÉ')); print(productos_por_agotarse(5))"

"""Pruebas funcionales del contrato de StockBridge."""

import json
import os
import tempfile
import unittest
from pathlib import Path

from mcp.server.mcpserver.exceptions import ToolError

import server


class PruebasInventario(unittest.TestCase):

    def test_busqueda_simple(self):
        self.assertEqual(
            [p["id"] for p in server.buscar_producto("miel")["productos"]],
            ["P002"]
        )

    def test_mayusculas_y_espacios(self):
        resultado = server.buscar_producto("  CAFÉ  ")
        self.assertEqual(
            [p["id"] for p in resultado["productos"]],
            ["P001", "P005"]
        )
        self.assertEqual(resultado["total"], 2)

    def test_sin_coincidencias(self):
        self.assertEqual(
            server.buscar_producto("arroz"),
            {"productos": [], "total": 0}
        )

    def test_busqueda_vacia(self):
        for valor in ("", "   "):
            with self.subTest(valor=valor), self.assertRaisesRegex(
                ToolError, "no vacío"
            ):
                server.buscar_producto(valor)

    def test_umbral_exclusivo_y_ordenado(self):
        resultado = server.productos_por_agotarse(5)

        self.assertEqual(
            [p["id"] for p in resultado["productos"]],
            ["P002", "P005", "P001"]
        )

        self.assertEqual(
            (resultado["total"], resultado["limite"]),
            (3, 5)
        )

        self.assertNotIn(
            "P003",
            [p["id"] for p in resultado["productos"]]
        )

    def test_umbral_por_defecto_y_cero(self):
        self.assertEqual(
            server.productos_por_agotarse(),
            server.productos_por_agotarse(5)
        )

        self.assertEqual(
            server.productos_por_agotarse(0),
            {"productos": [], "total": 0, "limite": 0}
        )

    def test_limite_invalido(self):
        for valor in (-1, 2.5, True, False, "5"):
            with self.subTest(valor=valor), self.assertRaisesRegex(
                ToolError, "entero"
            ):
                server.productos_por_agotarse(valor)

    def test_no_se_modifica_inventario(self):
        ruta = Path(server.__file__).with_name("inventario.json")
        antes = ruta.read_bytes()

        server.buscar_producto("café")
        server.productos_por_agotarse(5)

        self.assertEqual(ruta.read_bytes(), antes)

    def test_carga_desde_otra_carpeta(self):
        actual = Path.cwd()

        with tempfile.TemporaryDirectory() as carpeta:
            try:
                os.chdir(carpeta)
                self.assertEqual(
                    len(server.cargar_inventario()),
                    5
                )
            finally:
                os.chdir(actual)

    def test_datos_invalidos(self):
        casos = [
            "{",
            "{}",
            '[{"id":"P001"}]',
            json.dumps([
                {
                    "id": "P001",
                    "nombre": "Miel",
                    "precio_cop": 1,
                    "stock": -1
                }
            ]),
            json.dumps([
                {
                    "id": "P001",
                    "nombre": "Miel",
                    "precio_cop": True,
                    "stock": 0
                }
            ]),
            json.dumps([
                {
                    "id": "P001",
                    "nombre": "Miel",
                    "precio_cop": 1,
                    "stock": 0
                }
            ] * 2)
        ]

        with tempfile.TemporaryDirectory() as carpeta:
            ruta = Path(carpeta) / "inventario.json"

            with self.assertRaisesRegex(
                ValueError, "No se pudo cargar"
            ):
                server.cargar_inventario(ruta)

            for contenido in casos:
                with self.subTest(contenido=contenido):
                    with self.assertRaises(ValueError):
                        ruta.write_text(
                            contenido,
                            encoding="utf-8"
                        )
                        server.cargar_inventario(ruta)


if __name__ == "__main__":
    unittest.main()

"""Integración real: lanza server.py como subprocess y habla MCP por stdio."""

import sys
import unittest
from pathlib import Path

from mcp import Client, StdioServerParameters


class PruebasMCP(unittest.IsolatedAsyncioTestCase):

    async def test_descubrimiento_y_consultas_stdio(self):
        servidor = Path(__file__).resolve().with_name("server.py")

        proceso = StdioServerParameters(
            command=sys.executable,
            args=[str(servidor)]
        )

        async with Client(proceso) as cliente:

            herramientas = await cliente.list_tools()

            self.assertEqual(
                {t.name for t in herramientas.tools},
                {"buscar_producto", "productos_por_agotarse"}
            )

            for herramienta in herramientas.tools:
                self.assertIsNotNone(herramienta.input_schema)
                self.assertIsNotNone(herramienta.output_schema)

            bajos = await cliente.call_tool(
                "productos_por_agotarse",
                {"limite": 5}
            )

            self.assertFalse(bajos.is_error)

            self.assertEqual(
                [
                    p["id"]
                    for p in bajos.structured_content["productos"]
                ],
                ["P002", "P005", "P001"]
            )

            self.assertEqual(
                bajos.structured_content["total"],
                3
            )

            buscados = await cliente.call_tool(
                "buscar_producto",
                {"nombre": "  CAFÉ  "}
            )

            self.assertFalse(buscados.is_error)

            self.assertEqual(
                [
                    p["id"]
                    for p in buscados.structured_content["productos"]
                ],
                ["P001", "P005"]
            )

            error = await cliente.call_tool(
                "buscar_producto",
                {"nombre": "   "}
            )

            self.assertTrue(error.is_error)

            self.assertIn(
                "no vacío",
                str(error.content)
            )


if __name__ == "__main__":
    unittest.main()