MCP RAG

Um servidor gerenciado de Geração Aumentada por Recuperação (RAG) usando MCP, integrado com bases de conhecimento e OpenSearch.

GitHub
1
Experimente este MCPPatrocinado

Documentação

MCP RAG

Você pode usar RAG convenientemente usando MCP. Aqui, explicamos como usar MCP com o knowledge base, um serviço RAG totalmente gerenciado, e com o OpenSearch, um RAG gerenciado, além de como usar o advanced RAG, uma técnica para melhorar o desempenho do RAG. A arquitetura geral é mostrada abaixo e descreve 4 tipos de MCP que podem ser usados para aproveitar o RAG. Quando o usuário envia documentos para o Amazon S3, o Knowledge Base os importa por meio de sync e os carrega no Amazon OpenSearch Serverless. Nesse processo, o modelo de embedding predefinido é usado e as informações analisadas por meio de multimodal podem ser aproveitadas. Além disso, quando um documento é enviado ao Amazon S3, o evento gerado é recebido pelo AWS Lambda (s3-event-manager) e enviado ao SQS. Em seguida, o AWS Lambda (document-manager) carrega sequencialmente o contexto obtido por meio de embedding e análise multimodal no OpenSearch gerenciado. Os aplicativos de IA no Amazon EC2 podem usar as ferramentas do servidor MCP por meio da estrutura de cliente/servidor MCP. Ao usar RAG, você pode escolher um dos 4 métodos, conforme mostrado na figura abaixo: (1) AWS MCP (Knowledge Base), (2) MCP Lambda (Knowledge Base), (3) OpenSearch MCP e (4) MCP Lambda (OpenSearch). A infraestrutura necessária, como OpenSearch, Lambda e SQS, pode ser facilmente implantada usando AWS CDK.

image

Ao capturar cada página do documento e analisar as imagens, é possível obter mais informações sobre tabelas ou imagens dentro do documento. Para extrair e analisar imagens de páginas do documento, criamos e usamos um pipeline de processamento de dados em formato de evento, como em Implementando RAG Agêntico. Nesse caso, o processamento de eventos é feito por lambda-s3-event-manager e o processamento de documentos é implementado por lambda-document-manager.

image

Técnicas de Advanced RAG

Explicamos as técnicas de advanced RAG para melhorar o desempenho do RAG.

OCR

Após converter cada página do documento em imagem, analisamos por meio de multimodal. Nesse processo, usamos contextual embedding para uma análise de imagem adequada ao contexto. Para obter detalhes do código, consulte lambda-document-manager. Para contextual embedding, usamos o OpenSearch gerenciado.

O os_client é definido conforme abaixo e usado ao criar o índice do OpenSearch.

session = boto3.Session(region_name=region)
credentials = session.get_credentials()

awsauth = AWS4Auth(
    credentials.access_key,
    credentials.secret_key,
    region,
    'es',  
    session_token=credentials.token
)

os_client = OpenSearch(
    hosts=[{
        'host': opensearch_url.replace("https://", ""), 
        'port': 443
    }],
    http_compress=True,
    http_auth=awsauth,
    use_ssl=True,
    verify_certs=True,
    ssl_assert_hostname=False,
    ssl_show_warn=False,
    connection_class=RequestsHttpConnection
)

Agora definimos o vectorstore e o usamos ao adicionar ou excluir documentos.

from langchain_community.vectorstores.opensearch_vector_search import OpenSearchVectorSearch
vectorstore = OpenSearchVectorSearch(
    index_name=index_name,  
    is_aoss = False,
    embedding_function=bedrock_embeddings,
    opensearch_url=opensearch_url,
    http_auth=awsauth,
    connection_class=RequestsHttpConnection
)

Extraímos e usamos o contextual_text para entender o significado de cada página no documento completo.

def get_contextual_text(whole_text, splitted_text, llm): 
    contextual_template = (
        "<document>"
        "{WHOLE_DOCUMENT}"
        "</document>"
        "Here is the chunk we want to situate within the whole document."
        "<chunk>"
        "{CHUNK_CONTENT}"
        "</chunk>"
        "Please give a short succinct context to situate this chunk within the overall document for the purposes of improving search retrieval of the chunk."
        "Answer only with the succinct context and nothing else in English."
        "Put it in <result> tags."
    )              
    contextual_prompt = ChatPromptTemplate([
        ('human', contextual_template)
    ])

    contextual_text = ""        
    contexual_chain = contextual_prompt | llm            
    response = contexual_chain.invoke(
        {
            "WHOLE_DOCUMENT": whole_text,
            "CHUNK_CONTENT": splitted_text
        }
    )    
    output = response.content
    return output[output.find('<result>')+8:output.find('</result>')]

Para extrair texto das imagens das páginas, ajustamos o tamanho da imagem e a registramos no OpenSearch junto com o texto contextual.

def store_image_for_opensearch(key):
    image_obj = s3_client.get_object(Bucket=s3_bucket, Key=key)
    image_content = image_obj['Body'].read()
    img = Image.open(BytesIO(image_content))
                        
    width, height = img.size 
    print(f"width: {width}, height: {height}, size: {width*height}")
            
    isResized = False
    while(width*height > 5242880):
        width = int(width/2)
        height = int(height/2)
        isResized = True
        print(f"width: {width}, height: {height}, size: {width*height}")
           
    buffer = BytesIO()
    img.save(buffer, format="PNG")
    img_base64 = base64.b64encode(buffer.getvalue()).decode("utf-8")
                                                            
    llm = get_model()
    text = extract_text(llm, img_base64)
    extracted_text = text[text.find('<result>')+8:text.find('</result>')] 
    
    contextual_text = object_meta["contextual_text"]
    summary = summary_image(llm, img_base64, contextual_text)
    image_summary = summary[summary.find('<result>')+8:summary.find('</result>')]    
    contents = f"[이미지 요약]\n{image_summary}\n\n[추출된 텍스트]\n{extracted_text}"
    
    page = object_meta["page"]
    docs = []
    docs.append(
        Document(
            page_content=contents,
            metadata={
                'name': key,
                'page': page,
                'url': path+parse.quote(key)
            }
        )
    )
    return add_to_opensearch(docs)

Aqui, a função que extrai texto de imagens usando multimodal é a seguinte.

def summary_image(llm, img_base64, contextual_text):  
    query = "이미지가 의미하는 내용을 풀어서 자세히 알려주세요. markdown 포맷으로 답변을 작성합니다."
    if contextual_text:
        query += "\n아래 <reference>는 이미지와 관련된 내용입니다. 이미지 분석시 참고하세요. \n<reference>\n"+contextual_text+"\n</reference>"    
    messages = [
        HumanMessage(
            content=[
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/png;base64,{img_base64}", 
                    },
                },
                {
                    "type": "text", "text": query
                },
            ]
        )
    ]    
    result = llm.invoke(messages)
    extracted_text = result.content        
    return extracted_text

Parent Child Chunking

Para aumentar a precisão da busca de documentos e usar contexto suficiente, é necessário aplicar o Parent Child Chunking. Para isso, dividimos parent e child usando RecursiveCharacterTextSplitter, conforme abaixo.

parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=2000,
    chunk_overlap=100,
    separators=["\n\n", "\n", ".", " ", ""],
    length_function = len,
)
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=400,
    chunk_overlap=50,
    length_function = len,
)

Primeiro, inserimos os parent chunks no OpenSearch e verificamos os IDs. Adicionamos o ID do parent chunk ao meta do child chunk para que a busca seja feita pelo child chunk, mas o contexto real usa o texto do parent consultado pelo ID do parent. Como o texto contextual deve descrever as características do parent chunk que será realmente usado, adicionamos o texto contextual obtido do parent chunk ao child chunk. Em seguida, registramos também os child chunks no OpenSearch e verificamos os IDs. Salvamos os IDs de parent/child no meta do arquivo e os usamos ao atualizar/excluir documentos.

splitted_docs = parent_splitter.split_documents(docs)
parent_docs, contexualized_chunks = get_contextual_docs_using_parallel_processing(docs[-1], splitted_docs)

for i, doc in enumerate(parent_docs):
    doc.metadata["doc_level"] = "parent"
        
parent_doc_ids = vectorstore.add_documents(parent_docs, bulk_size = 10000)
ids = parent_doc_ids

for i, doc in enumerate(splitted_docs):
    _id = parent_doc_ids[i]
    child_docs = child_splitter.split_documents([doc])
    for _doc in child_docs:
        _doc.metadata["parent_doc_id"] = _id
        _doc.metadata["doc_level"] = "child"

    contexualized_child_docs = []
    for _doc in child_docs:
        contexualized_child_docs.append(
            Document(
                page_content=contexualized_chunks[i]+"\n\n"+_doc.page_content,
                metadata=_doc.metadata
            )
        )
    child_docs = contexualized_child_docs

    child_doc_ids = vectorstore.add_documents(child_docs, bulk_size = 10000)        
    ids += child_doc_ids

Uso do Knowledge Base

O knowledge base, um serviço RAG totalmente gerenciado, oferece suporte a sync para storages como S3, o que o torna conveniente de usar. No entanto, ao usar OCR ou contextual embedding, é necessário fazer parsing manualmente e inserir os dados usando Lambda ou outras ferramentas personalizadas.

Usando RAG com MCP

AWS MCP (Knowledge Base)

É possível consultar documentos do Amazon Knowledge Base usando o MCP fornecido pela AWS, como o Amazon Bedrock Knowledge Base Retrieval MCP Server. Nesse caso, não é necessária infraestrutura adicional para consulta, então a implementação é simples. A configuração do MCP nesse caso é a seguinte.

{
  "mcpServers": {
    "awslabs.bedrock-kb-retrieval-mcp-server": {
      "command": "uvx",
      "args": ["awslabs.bedrock-kb-retrieval-mcp-server@latest"],
      "env": {
        "AWS_PROFILE": "your-profile-name",
        "AWS_REGION": "us-east-1",
        "FASTMCP_LOG_LEVEL": "ERROR",
        "KB_INCLUSION_TAG_KEY": "optional-tag-key-to-filter-kbs",
        "BEDROCK_KB_RERANKING_ENABLED": "false"
      },
      "disabled": false,
      "autoApprove": []
    }
  }
}

O AWS knowledge base MCP consulta o knowledge base, encontra um knowledge base com uma tag específica (padrão: mcp-tag) e executa a query. Portanto, ao criar o knowledge base, é necessário configurar a tag conforme abaixo.

image

No Amazon Bedrock Knowledge Base Retrieval MCP Server, GetKnowledgeBases é chamado como resource, mas há um problema em que ele não é referenciado no LangGraph. Portanto, aqui copiamos o código e o modificamos como tool, conforme mcp_server_knowledge_base.py. A configuração do MCP nesse caso é a seguinte.

{
    "mcpServers": {
        "knowledge_base_custom": {
            "command": "python",
            "args": [
                "application/mcp_server_knowledge_base.py"
            ],
            "env": {
                "KB_INCLUSION_TAG_KEY": "mcp-rag"
            }
        }
    }
}

MCP Lambda (Knowledge Base)

Para consultar o knowledge base via MCP, usar Lambda permite implementar comportamentos de RAG personalizados de acordo com o objetivo do usuário. Abaixo, explicamos como definir um servidor MCP personalizado usando Lambda. O código relacionado à consulta do knowledge base via Lambda está em lambda-knowledge-base. Definimos knowledge_base_search como tool, conforme abaixo.

@mcp.tool()
def knowledge_base_search(keyword: str) -> list:
    """
    Search the knowledge base with the given keyword.
    keyword: the keyword to search
    return: the result of search
    """

    return rag.retrieve_knowledge_base(keyword)

O knowledge_base_search consulta documentos do knowledge base chamando diretamente o Lambda, como em retrieve_knowledge_base definido em mcp_knowledge_base.py.

def retrieve_knowledge_base(query):
    lambda_client = boto3.client(
        service_name='lambda',
        region_name=bedrock_region
    )
    functionName = f"knowledge-base-for-{projectName}"

    mcp_env = utils.load_mcp_env()
    grading_mode = mcp_env['grading_mode']
    multi_region = mcp_env['multi_region']

    payload = {
        'function': 'search_rag',
        'knowledge_base_name': knowledge_base_name,
        'keyword': query,
        'top_k': numberOfDocs,
        'grading': grading_mode,
        'model_name': model_name,
        'multi_region': multi_region
    }

    output = lambda_client.invoke(
        FunctionName=functionName,
        Payload=json.dumps(payload),
    )
    payload = json.load(output['Payload'])        
    return payload['response']

Implementar o servidor MCP com Lambda exige infraestrutura adicional, mas permite realizar tarefas personalizadas, como excluir documentos de baixa relevância por meio de grading, e obter respostas mais rápidas, pois a query é feita diretamente sem consultar a lista de knowledge bases.

OpenSearch MCP

Usando o OpenSearch MCP, é possível consultar o OpenSearch diretamente, sem recursos adicionais. No entanto, em junho de 2025, apenas a busca por texto é suportada, o que limita o desempenho. Ao usar o OpenSearch MCP, utilize a seguinte configuração.

{
    "mcpServers": {
        "opensearch-mcp-server": {
            "command": "uvx",
            "args": [
                "opensearch-mcp-server-py"
            ],
            "env": {
                "OPENSEARCH_URL": managed_opensearch_url,
                "AWS_REGION": aws_region,
                "AWS_ACCESS_KEY_ID": credentials.access_key,
                "AWS_SECRET_ACCESS_KEY": credentials.secret_key
            }
        }
    }
}

MCP Lambda (OpenSearch)

Ao implementar RAG personalizado, como OCR e contextual embedding, é necessário inserir dados diretamente no OpenSearch e consultá-los. Para consulta, você pode definir um servidor MCP personalizado com Lambda ou usar o OpenSearch MCP. A consulta ao OpenSearch via Lambda pode ser implementada como em lambda-opensearch. Para implementá-la como um servidor MCP personalizado, implemente conforme mcp_server_lambda_opensearch.py.

@mcp.tool()
def opensearch_search(keyword: str) -> list:
    """
    Search the knowledge base with the given keyword.
    keyword: the keyword to search
    return: the result of search
    """

    return rag.retrieve_opensearch(keyword)

Aqui, como em mcp_opensearch.py, você pode fazer uma solicitação direta a lambda-opensearch e obter os documentos do OpenSearch recuperados pelo Lambda. No Lambda, além de consultar documentos no OpenSearch, é possível realizar grading com base na relevância dos documentos.

def retrieve_opensearch(query):
    lambda_client = boto3.client(
        service_name='lambda',
        region_name=bedrock_region
    )
    functionName = f"opensearch-for-{projectName}"

    mcp_env = utils.load_mcp_env()
    grading_mode = mcp_env['grading_mode']
    multi_region = mcp_env['multi_region']

    payload = {
        'function': 'search_rag',
        'keyword': query,
        'top_k': numberOfDocs,
        'grading': grading_mode,
        'model_name': model_name,
        'multi_region': multi_region
    }
    output = lambda_client.invoke(
        FunctionName=functionName,
        Payload=json.dumps(payload),
    )
    payload = json.load(output['Payload'])        
    return payload['response']

AgentCore Gateway

O AgentCore Gateway pode fornecer servidores MCP via Lambda. Copie config.json.sample e crie o arquivo config.json.

cd gateway/opensearch-retriever && cp config.json.sample config.json

No config.json, defina o nome do projeto, a URL do OpenSearch e o sharing_url, que é o domínio do CloudFront, conforme abaixo. Essas informações podem ser verificadas nos outputs gerados na implantação do CDK.

{
  "projectName": "mcp-rag",
  "opensearch_url": "https://search-mcp-rag-mxtkul2z3qv5iiqprb7q3jx4wy.us-west-2.es.amazonaws.com",
  "sharing_url": "https://d20lfnyi6fvd87.cloudfront.net"
}

Instale a role para o gateway chamado "mcp-rag" no AgentCore da seguinte forma.

python create_gateway_role.py

O resultado nesse caso é o seguinte.

image

Instale o AgentCore Gateway conforme abaixo e implante "opensearch-retriever" como target. Se não houver um Lambda RAG para consultar o OpenSearch, instale-o.

python create_gateway_tool.py

A instalação foi concluída, então teste o funcionamento conforme abaixo.

test_mcp_remote.py

O resultado nesse caso é o seguinte. Se o bearer token do secret expirar, renove-o e tente acessar novamente. Nesse momento, use list_tools para verificar as ferramentas disponíveis e depois execute.

image

Instalação

Clone o repositório.

git clone https://github.com/kyopark2014/mcp-rag/

Instale as bibliotecas necessárias.

cd mcp-rag && pip install -r requirements.txt

Instale os secrets de infraestrutura necessária para o CDK, como CloudFront, S3, OpenSearch, Knowledge base, tavily e weather. Se o cdk bootstrap não estiver configurado, instale-o e depois execute.

cd cdk-mcp-rag/ && cdk deploy --all

Após a instalação, copie "CdkMcpRagStack.environmentformcprag" conforme abaixo e crie o arquivo application/config.json.

image

O config.json contém as informações necessárias para o funcionamento do agent e não é compartilhado via git, conforme .gitignore. Um exemplo do config.json gerado é o seguinte.

{
    "projectName":"mcp-rag",
    "accountId":"862926741992",
    "region":"us-west-2",
    "roleKnowledgeBase":"arn:aws:iam::862926741992:role/role-knowledge-base-for-mcp-rag-us-west-2",
    "collectionArn":"arn:aws:aoss:us-west-2:862926741992:collection/8krsnuq4it9gpl70i3u6",
    "serverless_opensearch_url":"https://8krsnuq4it9gpl70i3u6.us-west-2.aoss.amazonaws.com",
    "managed_opensearch_url":"https://search-mcp-rag-mxtkul3z3qv5iiqprb7q3jx4wy.us-west-2.es.amazonaws.com",
    "knowledge_base_role":"arn:aws:iam::862926741992:role/role-knowledge-base-for-mcp-rag-us-west-2",
    "s3_bucket":"storage-for-mcp-rag-862926741992-us-west-2",
    "s3_arn":"arn:aws:s3:::storage-for-mcp-rag-862926741992-us-west-2",
    "sharing_url":"https://d3mo4kqj5cjiuy.cloudfront.net"
 }

Em seguida, acesse o Secret Manager e insira as credenciais conforme abaixo.

image

Se o streamlit não estiver instalado, consulte streamlit para instalá-lo. Depois, execute conforme abaixo.

streamlit run application/app.py

Ao executar, a tela abaixo é exibida. Selecione o Agent para executá-lo e verificar o funcionamento.

Resultados da execução

Aqui, fornecemos os seguintes servidores MCP.

image

Se você selecionar "AWS MCP (Knowledge Base)", o GetKnowledgeBases busca o knowledge base com a tag mcp-rag e, em seguida, o QueryKnowledgeBases executa a busca, conforme abaixo.

Se você selecionar "MCP Lambda (Knowledge Base)" e pesquisar, poderá obter resultados de busca de documentos usando knowledge_base_search, conforme abaixo.

Se você pesquisar com "OpenSearch MCP", a consulta é feita usando o OpenSearch MCP. Como é uma busca por texto, pode não retornar resultados, conforme abaixo.

Se você pesquisar "código boiler?" no "OpenSearch MCP", a busca é bem-sucedida, conforme abaixo.

Se você pesquisar com "MCP Lambda (OpenSearch)", poderá obter resultados da busca no OpenSearch usando opensearch_search, conforme abaixo.

Referência

Por que o RAG empresarial falha… Como escalar o RAG em ambientes corporativos