Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 15 additions & 12 deletions modelcontextprotocol/.cursor/rules/project-structure.mdc
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,13 @@ The main entry point for the MCP server. Registers and exposes tools defined in
```python
from mcp.server.fastmcp import FastMCP
from tools import (
search_assets,
get_assets_by_dsl,
traverse_lineage,
update_assets,
search_assets as search_assets_impl,
get_assets_by_dsl as get_assets_by_dsl_impl,
traverse_lineage as traverse_lineage_impl,
update_assets as update_assets_impl,
create_glossary_category_assets as create_glossary_category_assets_impl,
create_glossary_assets as create_glossary_assets_impl,
create_glossary_term_assets as create_glossary_term_assets_impl,
UpdatableAttribute,
CertificateStatus,
UpdatableAsset,
Expand All @@ -51,35 +54,35 @@ mcp = FastMCP("Atlan MCP", dependencies=["pyatlan"])
# Refer to the actual server.py and tools implementations for details.

@mcp.tool()
def search_assets_tool(
def search_assets(
conditions: Optional[Union[Dict[str, Any], str]] = None,
# ... many other parameters ...
):
"""Advanced asset search using FluentSearch with flexible conditions."""
return search_assets(conditions=conditions, ...)
return search_assets_impl(conditions=conditions, ...)

@mcp.tool()
def get_assets_by_dsl_tool(dsl_query: Union[str, Dict[str, Any]]):
def get_assets_by_dsl(dsl_query: Union[str, Dict[str, Any]]):
"""Execute the search with the given DSL query."""
return get_assets_by_dsl(dsl_query)
return get_assets_by_dsl_impl(dsl_query)

@mcp.tool()
def traverse_lineage_tool(
def traverse_lineage(
guid: str,
direction: str, # UPSTREAM or DOWNSTREAM
# ... other parameters ...
):
"""Traverse asset lineage in specified direction."""
return traverse_lineage(guid=guid, direction=direction, ...)
return traverse_lineage_impl(guid=guid, direction=direction, ...)

@mcp.tool()
def update_assets_tool(
def update_assets(
assets: Union[UpdatableAsset, List[UpdatableAsset]],
attribute_name: UpdatableAttribute,
attribute_values: List[Union[CertificateStatus, str]],
):
"""Update one or multiple assets with different values for the same attribute."""
return update_assets(assets=assets, attribute_name=attribute_name, attribute_values=attribute_values)
return update_assets_impl(assets=assets, attribute_name=attribute_name, attribute_values=attribute_values)
```

### settings.py
Expand Down
18 changes: 9 additions & 9 deletions modelcontextprotocol/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -173,7 +173,7 @@ You can restrict access to specific tools using the `RESTRICTED_TOOLS` environme
"-e",
"ATLAN_AGENT_ID=<YOUR_AGENT_ID>",
"-e",
"RESTRICTED_TOOLS=get_assets_by_dsl_tool,update_assets_tool",
"RESTRICTED_TOOLS=get_assets_by_dsl,update_assets",
"ghcr.io/atlanhq/atlan-mcp-server:latest"
]
}
Expand All @@ -193,7 +193,7 @@ You can restrict access to specific tools using the `RESTRICTED_TOOLS` environme
"ATLAN_API_KEY": "<YOUR_API_KEY>",
"ATLAN_BASE_URL": "https://<YOUR_INSTANCE>.atlan.com",
"ATLAN_AGENT_ID": "<YOUR_AGENT_ID>",
"RESTRICTED_TOOLS": "get_assets_by_dsl_tool,update_assets_tool"
"RESTRICTED_TOOLS": "get_assets_by_dsl,update_assets"
}
}
}
Expand All @@ -204,10 +204,10 @@ You can restrict access to specific tools using the `RESTRICTED_TOOLS` environme

You can restrict any of the following tools:

- `search_assets_tool` - Asset search functionality
- `get_assets_by_dsl_tool` - DSL query execution
- `traverse_lineage_tool` - Lineage traversal
- `update_assets_tool` - Asset updates (descriptions, certificates)
- `search_assets` - Asset search functionality
- `get_assets_by_dsl` - DSL query execution
- `traverse_lineage` - Lineage traversal
- `update_assets` - Asset updates (descriptions, certificates)
- `create_glossaries` - Glossary creation
- `create_glossary_categories` - Category creation
- `create_glossary_terms` - Term creation
Expand All @@ -217,19 +217,19 @@ You can restrict any of the following tools:
#### Read-Only Access
Restrict all write operations:
```
RESTRICTED_TOOLS=update_assets_tool,create_glossaries,create_glossary_categories,create_glossary_terms
RESTRICTED_TOOLS=update_assets,create_glossaries,create_glossary_categories,create_glossary_terms
```

#### Disable DSL Queries
For security or performance reasons:
```
RESTRICTED_TOOLS=get_assets_by_dsl_tool
RESTRICTED_TOOLS=get_assets_by_dsl
```

#### Minimal Access
Allow only basic search:
```
RESTRICTED_TOOLS=get_assets_by_dsl_tool,update_assets_tool,traverse_lineage_tool,create_glossaries,create_glossary_categories,create_glossary_terms
RESTRICTED_TOOLS=get_assets_by_dsl,update_assets,traverse_lineage,create_glossaries,create_glossary_categories,create_glossary_terms
```

### How It Works
Expand Down
44 changes: 22 additions & 22 deletions modelcontextprotocol/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,13 @@
from typing import Any, Dict, List
from fastmcp import FastMCP
from tools import (
search_assets,
get_assets_by_dsl,
traverse_lineage,
update_assets,
create_glossary_category_assets,
create_glossary_assets,
create_glossary_term_assets,
search_assets as search_assets_impl,
get_assets_by_dsl as get_assets_by_dsl_impl,
traverse_lineage as traverse_lineage_impl,
update_assets as update_assets_impl,
create_glossary_category_assets as create_glossary_category_assets_impl,
create_glossary_assets as create_glossary_assets_impl,
create_glossary_term_assets as create_glossary_term_assets_impl,
UpdatableAttribute,
CertificateStatus,
UpdatableAsset,
Expand Down Expand Up @@ -40,7 +40,7 @@


@mcp.tool()
def search_assets_tool(
def search_assets(
conditions=None,
negative_conditions=None,
some_conditions=None,
Expand Down Expand Up @@ -242,7 +242,7 @@ def search_assets_tool(
domain_guids = parse_list_parameter(domain_guids)
guids = parse_list_parameter(guids)

return search_assets(
return search_assets_impl(
conditions,
negative_conditions,
some_conditions,
Expand All @@ -266,7 +266,7 @@ def search_assets_tool(


@mcp.tool()
def get_assets_by_dsl_tool(dsl_query):
def get_assets_by_dsl(dsl_query):
"""
Execute the search with the given query
dsl_query : Union[str, Dict[str, Any]] (required):
Expand Down Expand Up @@ -334,11 +334,11 @@ def get_assets_by_dsl_tool(dsl_query):
}'''
response = get_assets_by_dsl(dsl_query)
"""
return get_assets_by_dsl(dsl_query)
return get_assets_by_dsl_impl(dsl_query)


@mcp.tool()
def traverse_lineage_tool(
def traverse_lineage(
guid,
direction,
depth=1000000,
Expand Down Expand Up @@ -374,7 +374,7 @@ def traverse_lineage_tool(

Examples:
# Get lineage with default attributes
lineage = traverse_lineage_tool(
lineage = traverse_lineage(
guid="asset-guid-here",
direction="DOWNSTREAM",
depth=1000,
Expand All @@ -391,7 +391,7 @@ def traverse_lineage_tool(
# Parse include_attributes parameter if provided
parsed_include_attributes = parse_list_parameter(include_attributes)

return traverse_lineage(
return traverse_lineage_impl(
guid=guid,
direction=direction_enum,
depth=int(depth),
Expand All @@ -402,7 +402,7 @@ def traverse_lineage_tool(


@mcp.tool()
def update_assets_tool(
def update_assets(
assets,
attribute_name,
attribute_values,
Expand All @@ -426,7 +426,7 @@ def update_assets_tool(

Examples:
# Update certificate status for a single asset
result = update_assets_tool(
result = update_assets(
assets={
"guid": "asset-guid-here",
"name": "Asset Name",
Expand All @@ -438,7 +438,7 @@ def update_assets_tool(
)

# Update user description for multiple assets
result = update_assets_tool(
result = update_assets(
assets=[
{
"guid": "asset-guid-1",
Expand All @@ -460,7 +460,7 @@ def update_assets_tool(
)

# Update readme for a single asset with Markdown
result = update_assets_tool(
result = update_assets(
assets={
"guid": "asset-guid-here",
"name": "Asset Name",
Expand Down Expand Up @@ -496,7 +496,7 @@ def update_assets_tool(
else:
updatable_assets = [UpdatableAsset(**asset) for asset in parsed_assets]

return update_assets(
return update_assets_impl(
updatable_assets=updatable_assets,
attribute_name=attr_enum,
attribute_values=parsed_attribute_values,
Expand Down Expand Up @@ -557,7 +557,7 @@ def create_glossaries(glossaries) -> List[Dict[str, Any]]:
except json.JSONDecodeError as e:
return {"error": f"Invalid JSON format for glossaries parameter: {str(e)}"}

return create_glossary_assets(glossaries)
return create_glossary_assets_impl(glossaries)


@mcp.tool()
Expand Down Expand Up @@ -622,7 +622,7 @@ def create_glossary_terms(terms) -> List[Dict[str, Any]]:
except json.JSONDecodeError as e:
return {"error": f"Invalid JSON format for terms parameter: {str(e)}"}

return create_glossary_term_assets(terms)
return create_glossary_term_assets_impl(terms)


@mcp.tool()
Expand Down Expand Up @@ -691,7 +691,7 @@ def create_glossary_categories(categories) -> List[Dict[str, Any]]:
except json.JSONDecodeError as e:
return {"error": f"Invalid JSON format for categories parameter: {str(e)}"}

return create_glossary_category_assets(categories)
return create_glossary_category_assets_impl(categories)


def main():
Expand Down