Extract data from legacy SAP R/3 systems and convert it into modern REST APIs for AI agent consumption. This tool bridges the gap between enterprise SAP systems and modern AI applications.
- Universal Tool Interface: Implements the Universal LLM Agent Tool standard
- SAP Data Extraction: Extract orders, customers, materials, warehouses, and inventory
- RFC Connectivity: Direct connection to SAP using Remote Function Calls
- Data Transformation: Automatic conversion from SAP format to JSON
- Multi-Framework Support: Works with LangChain, CrewAI, AutoGen, MCP, and more
- Caching: Built-in caching for improved performance
- Async Support: Handle large data extractions asynchronously
- Security: Secure credential handling and data encryption
- Python 3.9+
- SAP R/3 system (4.6C or higher)
- SAP RFC SDK (SAP NetWeaver RFC SDK)
- Valid SAP user credentials with appropriate permissions
First, download and install the SAP NetWeaver RFC SDK from the SAP Service Marketplace.
# Set environment variables
export SAPNWRFC_HOME=/path/to/sap/nwrfcsdk
export LD_LIBRARY_PATH=$SAPNWRFC_HOME/lib:$LD_LIBRARY_PATHpip install sap-r3-integration-toolOr install from source:
git clone https://github.com/KazKozDev/sap_r3_integration_tool.git
cd sap_r3_integration_tool
pip install -e .from sap_r3_integration import SAPR3IntegrationTool, ToolConfig
# Configure the tool
config = ToolConfig(
sap_config={
"ashost": "sap.example.com",
"sysnr": "00",
"client": "100",
"user": "SAP_USER",
"passwd": "SAP_PASSWORD"
}
)
# Create tool instance
tool = SAPR3IntegrationTool(config)
# Extract customer data
result = tool.execute({
"data_type": "customers",
"filters": {"LAND1": "US"},
"limit": 100
})
print(f"Extracted {result['metadata']['count']} customers")from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI
# Convert to LangChain tool
langchain_tool = tool.to_langchain()
# Create agent
llm = OpenAI(temperature=0)
agent = initialize_agent(
[langchain_tool],
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION
)
# Use the tool
response = agent.run("Get all customers from Germany")# Convert to MCP format
mcp_tool = tool.to_mcp_server()
# Register with MCP server
mcp_server.register_tool(mcp_tool)| Data Type | Description | SAP Tables |
|---|---|---|
orders |
Sales orders with items | VBAK, VBAP |
customers |
Customer master data | KNA1 |
materials |
Material master data | MARA, MAKT |
warehouses |
Storage locations | T001L |
inventory |
Current stock levels | MARD |
# Date range filter
orders = tool.execute({
"data_type": "orders",
"filters": {
"ERDAT": {
"from": "20250101",
"to": "20250131"
}
}
})
# Multiple filters
inventory = tool.execute({
"data_type": "inventory",
"filters": {
"WERKS": "1000", # Plant
"LGORT": "0001" # Storage location
}
})import asyncio
async def extract_large_dataset():
result = await tool.execute_async({
"data_type": "orders",
"limit": 10000
})
return result
# Run async extraction
data = asyncio.run(extract_large_dataset())# Extract specific fields only
customers = tool.execute({
"data_type": "customers",
"fields": ["KUNNR", "NAME1", "LAND1"],
"limit": 50
})Never hardcode credentials. Use environment variables or secure vaults:
import os
from sap_r3_integration import SAPR3IntegrationTool, ToolConfig
config = ToolConfig(
sap_config={
"ashost": os.getenv("SAP_HOST"),
"client": os.getenv("SAP_CLIENT"),
"user": os.getenv("SAP_USER"),
"passwd": os.getenv("SAP_PASSWORD")
}
)The SAP user needs the following authorizations:
S_RFC: RFC accessS_TABU_DIS: Table display authorization- Access to specific tables (VBAK, KNA1, etc.)
┌─────────────────┐ ┌──────────────┐ ┌─────────────┐
│ AI Agent │────▶│ Tool Layer │────▶│ SAP R/3 │
│ (LangChain, │ │ │ │ │
│ CrewAI, etc) │ │ - Validate │ │ - RFC Call │
└─────────────────┘ │ - Transform │ │ - Tables │
│ - Cache │ └─────────────┘
└──────────────┘
Configure caching for frequently accessed data:
from sap_r3_integration import ToolConfig, CacheConfig
config = ToolConfig(
cache_config=CacheConfig(
backend="redis",
ttl=3600, # 1 hour
max_size=1000
)
)The tool automatically manages connection pooling to optimize performance.
Run the test suite:
# Run all tests
pytest
# Run with coverage
pytest --cov=sap_r3_integration --cov-report=html
# Run specific test category
pytest tests/test_extraction.py-
RFC Connection Failed
Error: RFC_COMMUNICATION_FAILURE Solution: Check network connectivity and SAP system availability -
Authorization Error
Error: RFC_AUTHORIZATION_FAILURE Solution: Verify user permissions in SAP (transaction SU01) -
Data Extraction Timeout
Solution: Reduce limit or use async execution
Enable detailed logging:
import logging
logging.basicConfig(level=logging.DEBUG)class ToolConfig:
environment: str = "prod"
timeout: int = 30
retries: int = 3
cache_config: CacheConfig
sap_config: Dict[str, Any]{
"data_type": str, # Required: Type of data to extract
"filters": dict, # Optional: Filter criteria
"limit": int, # Optional: Max records (default: 100)
"fields": List[str] # Optional: Specific fields to extract
}{
"success": bool,
"data": List[Dict],
"metadata": {
"count": int,
"extraction_time": str,
"data_type": str,
"execution_time": float
},
"error": { # Only if success=false
"code": str,
"message": str
}
}We welcome contributions! Please follow these steps:
- Fork the repository
- Create your feature branch (
git checkout -b feature/amazing-feature) - Commit your changes (
git commit -m 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
This project is licensed under the MIT License.
- SAP NetWeaver RFC SDK team
- PyRFC maintainers
For support, please contact:
- Email: [email protected]
- GitHub Issues: Report an issue
- Support for SAP S/4HANA
- Real-time change data capture
- GraphQL API endpoint
- Built-in data validation rules
- Support for custom ABAP functions
Created and maintained by KazKozDev