Приложения для получения построения с достопримечательностью (RAG) стали краеугольным камнем современного ИИ, что позволяет крупным языковым моделям (LLMS) заземлять свои результаты в современных, специфических для домена знаниях. Традиционно, тряпичные трубопроводы полагались на встраиваемые и векторные базы данных, чтобы извлечь соответствующий контекст во время вывода. Тем не менее, по мере того, как системы ИИ растут более сложными, а спрос на защищенную и масштабируемую интеграцию в реальном времени повышается, протокол контекста модели (MCP) становится преобразующим стандартом-не только для вызова инструментов, но и для всей экосистемы RAG.
Ресурсы в MCP представляют собой только для чтения, адресуемые объекты данных, выявленные серверами, такими как файлы, записи базы данных, ответы API или даже статистику рынка в реальном времени. В отличие от инструментов, которые о том, что может сделать ИИ », ресурсы касаются« о том, что должен знать ИИ » — они предоставляют структурированные контекстные данные непосредственно в рабочую память модели, готовые к суммированию, анализу или фактическим заземлениям. Это означает, что приложения RAG, построенные на MCP, могут обойти накладные расходы по встраиванию и поиска вектора — извлечение живой, авторитетной информации о спросе (и всегда работая с самым свежим контекстом). Это особенно важно при работе с данными, хранящимися в структурированном формате, например, в базах данных реляционных и NOSQL.
Что делает ресурсы MCP особенно убедительными для RAG, так это их гибкость и обнаружение. Серверы MCP могут подвергать статические или динамические ресурсы, поддерживать иерархические данные и даже аннотировать ресурсы метаданными для более разумного отбора и приоритетов. Клиенты могут обнаружить доступные ресурсы во время выполнения, выбирать, что имеет значение, и получить их стандартизированным, безопасным способом — обеспечивая более прозрачные рабочие процессы RAG, которые являются более прозрачными, проверенными и адаптируемыми, чем когда -либо прежде.
Этот урок покажет, как объединить ресурсы MCP с инструментами для реализации тряпичного и инструментального вызова. Ключевым моментом, который следует помнить, является то, что ресурсы MCP предназначены для доступа к данным в форме только для чтения, в то время как инструменты MCP предназначены для выполнения действий, которые манипулируют данными.
Давайте начнем с заполнения базы данных SQLITE с помощью образцов данных сотрудника. Это будет действовать как хранилище данных для нашего MCP -сервера.
Import SQLite3 Import OS os # создать каталог DB, если он не существует, OS.MadkedIrs (‘DB’, ESIGE_OK = TRUE) # Подключитесь к базе данных SQLITE (создает его, если он не существует) conn = sqlite3.connect (‘db/сотрудники. First_name Text не null, последний текст, не нулевой, текст электронной почты уникальный, текст отдела, зарплата реальная, Hire_date Text) » ‘) # Образец Data Data Data Mecorments = [
(1, ‘John’, ‘Doe’, ‘john.doe@company.com’, ‘Engineering’, 85000, ‘2020-01-15’),
(2, ‘Jane’, ‘Smith’, ‘jane.smith@company.com’, ‘Marketing’, 78000, ‘2019-03-20’),
(3, ‘Michael’, ‘Johnson’, ‘michael.j@company.com’, ‘Sales’, 92000, ‘2021-06-10’),
(4, ‘Emily’, ‘Williams’, ’emily.w@company.com’, ‘HR’, 65000, ‘2022-02-28’),
(5, ‘David’, ‘Brown’, ‘david.b@company.com’, ‘Engineering’, 95000, ‘2018-11-05’),
(6, ‘Sarah’, ‘Davis’, ‘sarah.d@company.com’, ‘Finance’, 88000, ‘2021-09-15’),
(7, ‘Robert’, ‘Miller’, ‘robert.m@company.com’, ‘Sales’, 89000, ‘2020-07-22’),
(8, ‘Lisa’, ‘Wilson’, ‘lisa.w@company.com’, ‘Marketing’, 82000, ‘2019-12-01’),
(9, ‘James’, ‘Taylor’, ‘james.t@company.com’, ‘Engineering’, 91000, ‘2022-04-18’),
(10, ‘Jennifer’, ‘Anderson’, ‘jennifer.a@company.com’, ‘HR’, 68000, ‘2021-11-30’)
]
# Вставьте образцы данных cursor.executemany (» » вставить или заменить в сотрудники (id, first_name, last_name, электронная почта, отделение, зарплата, hire_date) значения (?,?,? 12345678910111213141516171819202122232425262728293031323334353637383940414243445464748 Import3import OS # Create DB, если он не существует. существует_ok = true) # подключиться к базе данных sqlite (создает ее, если она не существует) conn = sqlite3.conce (‘db/commerial.db’) cursor = conn.cursor () # Создание сотрудников Tablecursor.execute (» ‘Создать таблицу, если не существует, сотрудники (ID Integer Primary Primary Key, First_Nam Hire_date Text) » ‘) # Образец сотрудника DataMployee = [ (1, ‘John’, ‘Doe’, ‘john.doe@company.com’, ‘Engineering’, 85000, ‘2020-01-15’), (2, ‘Jane’, ‘Smith’, ‘jane.smith@company.com’, ‘Marketing’, 78000, ‘2019-03-20’), (3, ‘Michael’, ‘Johnson’, ‘michael.j@company.com’, ‘Sales’, 92000, ‘2021-06-10’), (4, ‘Emily’, ‘Williams’, ’emily.w@company.com’, ‘HR’, 65000, ‘2022-02-28’), (5, ‘David’, ‘Brown’, ‘david.b@company.com’, ‘Engineering’, 95000, ‘2018-11-05’), (6, ‘Sarah’, ‘Davis’, ‘sarah.d@company.com’, ‘Finance’, 88000, ‘2021-09-15’), (7, ‘Robert’, ‘Miller’, ‘robert.m@company.com’, ‘Sales’, 89000, ‘2020-07-22’), (8, ‘Lisa’, ‘Wilson’, ‘lisa.w@company.com’, ‘Marketing’, 82000, ‘2019-12-01’), (9, ‘James’, ‘Taylor’, ‘james.t@company.com’, ‘Engineering’, 91000, ‘2022-04-18’), (10, ‘Jennifer’, ‘Anderson’, ‘jennifer.a@company.com’, ‘HR’, 68000, ‘2021-11-30’)] # Вставьте пример datacursor.executemany (» » ‘Вставить или заменить в сотрудники (id, first_name, last_name, электронная почта, отделение, зарплата, hire_date) значения (?,?,?
Наша цель — создать сервер MCP, который может извлечь данные из этой базы данных, используя ресурсы и манипулировать данными, используя инструменты.
Давайте теперь создадим сервер MCP для разоблачения ресурсов и инструментов. Поскольку мы реализуем их как асинхронные методы, давайте установим aiosqlite и, конечно же, модуль Python для MCP, FASTMCP.
PIP установить FastMCP AIOSQLite 1 PIP установить FASTMCP AIOSQLITE
Во -первых, давайте добавим ресурсы на сервер MCP.
Импорт aiosqlite из fastmcp import fastmcp из списка импорта типа, DICT, необязательный MCP = FASTMCP («MCP -сервер сотрудников») db_path = «/db/employes.db» @mcp.resource («Сотрудники: // все») async def_all_empulitee () -& gt; Список[Dict]: «» «Возвращает все записи сотрудников в виде списка словарей.» «» Async with aiosqlite.connect (db_path) как conn: cursor = await conn.execut [column[0] для столбца в cursor.description]сотрудники = [dict(zip(columns, row)) async for row in cursor]
await cursor.close () возвращать сотрудники @mcp.resource («Сотрудники: // {employee_id}») async def get_employee_by_id (employee_id: int) -& gt; Необязательный[Dict]: «» «Возвращает единую запись сотрудника на основе идентификатора сотрудника.» «» Async с aiosqlite.connect (db_path) как conn: count await conn.execute (‘select * from hight where id =? [column[0] for column in cursor.description]result = dict(zip(columns, row)) else: result = None await cursor.close() return result 12345678910111213141516171819202122232425262728293031 import aiosqlitefrom fastmcp import FastMCPfrom typing import List, Dict, Необязательный MCP = FASTMCP («MCP -сервер сотрудников») db_path = «/db/employees.db» @mcp.resource («Сотрудники: // all») async def get_alleeeseees () -& gt; Список[Dict]: «» «Возвращает все записи сотрудников в виде списка словарей.» «» Async with aiosqlite.connect (db_path) как conn: cursor = await conn.execut [column[0] для столбца в cursor.description]сотрудники = [dict(zip(columns, row)) async for row in cursor] await cursor.close () возвращать сотрудники @mcp.resource («Сотрудники: // {employee_id}») async def get_employee_by_id (employee_id: int) -& gt; Необязательный[Dict]: «» «Возвращает единую запись сотрудника на основе идентификатора сотрудника.» «» Async с aiosqlite.connect (db_path) как conn: count await conn.execute (‘select * from hight where id =? [column[0] Для столбца в cursor.description]result = dict (Zip (столбцы, строка)) иначе: result = none a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a a ждать cursor.close ()
У нас есть две функции — get_all_employees () и get_employeee_by_id — чтобы получить все строки и строку от идентификатора сотрудника соответственно.
Обратите внимание, как мы аннотировали эти две функции. Первый — @mcp.resource («Сотрудники: // Все») — действует как прозвище для клиентов, чтобы получить все записи, в то время как вторая функция, аннотированная с @mcp.resource («Сотрудники: // {employeee_id}») принимает параметр, который является идентификатором сотрудника.
Клиенты MCP могут использовать эти ресурсы для получения контекста, который можно передать в LLMS, чтобы обосновать их ответ.
Тот же клиент MCP может захотеть манипулировать данными, обновляя и удаляя существующие записи. Это может использоваться агентом ИИ, который выполняет кадры. Давайте создадим инструмент MCP для удаления сотрудника из системы.
@mcp.tool () async def delete_employee (employee_id: int) -& gt; Bool: «» «Удаляет запись сотрудника на основе идентификатора сотрудника. Возвращает True, если успешно.» «» Async с aiosqlite.connect (db_path) как conn: try: coursor = wait conn.execute (‘Delete от сотрудников, где id =? 0 await cursor.close (), кроме исключения: успех = ложный успех возврата 123456789101112 @mcp.tool () async def delete_employee (employee_id: int) -& gt; Bool: «» «Удаляет запись сотрудника на основе идентификатора сотрудника. Возвращает True, если успешно.» «» Async с aiosqlite.connect (db_path) как conn: try: coursor = wait conn.execute (‘Delete от сотрудников, где id =? 0 ждать cursor.close () За исключением исключения: успех = ложный успех возврата
Пришло время написать основную функцию, которая действует как точка входа.
Если __name__ == «__main__»: mcp.run (transport = «stdio») 12, если __name__ == «__main__»: mcp.run (transport = «stdio»)
Запустите сервер MCP, чтобы сделать ресурсы и инструменты доступными для клиентов. Мы используем транспорт Stdio для простоты.
Пришло время написать клиенту, чтобы проверить, доступен ли сервер MCP, и разоблачает ожидаемые ресурсы и инструменты.
Импорт Asyncio из FASTMCP импортного клиента Async def main (): # подключиться к серверу через stdio (запустив Server.py в качестве подпроцесса) Async с клиентом («server.py») как клиент: print («resources:») resources = wait client.list_resources () prints) print («\ nlisting tools:») 12345678910111213 Импорт Asynciofrom Fastmcp импорт клиент Async def main (): # подключиться к серверу с помощью stdio (запустив Server.py как подпроцесс) Async с клиентом («Server.py») как клиент: printing («Пеллинг ресурсов:» resources = await client. client.list_tools () print (инструменты)
Если __name__ == «__main__»: asyncio.run (main ())
Если все идет хорошо, вы должны увидеть следующий вывод:
Поздравляю! Вы успешно создали MCP -сервер с комбинацией ресурсов и инструментов. В следующей части этой серии я продемонстрирую, как интегрировать это с помощью LLM, создав тряпное приложение, основанное на GPT-4.1. Следите за обновлениями.
Trending Stories youtube.com/thenewstack Tech движется быстро, не пропустите эпизод. Подпишитесь на наш канал YouTube, чтобы транслировать все наши подкасты, интервью, демонстрации и многое другое. Группа подпишитесь с эскизом. Janakiram MSV является основным аналитиком в Janakiram & Associates и адъюнкт -преподавателем Международного института информационных технологий. Он также является квалифицированным Google Cloud Developer, сертифицированным архитектором решений Amazon, сертифицированным разработчиком Amazon, … Подробнее от Janakiram MSV
